页面部分实现
This commit is contained in:
YueL1331
169
1.1/frps_final.py
Normal file
169
1.1/frps_final.py
Normal file
@ -0,0 +1,169 @@
|
||||
import requests
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
# --- 基础配置 ---
|
||||
BASE_URL = "http://106.75.72.40:7500/api/proxy/tcp"
|
||||
PRIMARY_AUTH = "Basic YWRtaW46bGljYWhr"
|
||||
LOGIN_PAYLOAD = {"username": "admin", "password": "licahk", "recaptcha": ""}
|
||||
|
||||
SAVE_DIR = "downloaded_data"
|
||||
if not os.path.exists(SAVE_DIR):
|
||||
os.makedirs(SAVE_DIR)
|
||||
|
||||
|
||||
def get_today_str():
|
||||
return datetime.now().strftime("%Y_%m_%d")
|
||||
|
||||
|
||||
def get_dynamic_token(port):
|
||||
"""
|
||||
为指定端口的站点执行登录，获取最新的 x-auth token
|
||||
"""
|
||||
login_url = f"http://106.75.72.40:{port}/api/login"
|
||||
try:
|
||||
# 登录不需要 x-auth，只需要 Basic Auth 或特定的 Payload
|
||||
resp = requests.post(login_url, json=LOGIN_PAYLOAD, timeout=10)
|
||||
if resp.status_code == 200:
|
||||
# 登录成功后，token 通常直接返回在响应体中（纯字符串或 JSON）
|
||||
token = resp.text.strip().replace('"', '')
|
||||
return token
|
||||
else:
|
||||
print(f" ❌ 登录失败: 端口 {port}, 状态码 {resp.status_code}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f" ❌ 登录异常: {port}, {e}")
|
||||
return None
|
||||
|
||||
|
||||
def find_closest_item(items, is_date_level=True):
|
||||
if not items or not isinstance(items, list): return None
|
||||
today = datetime.now()
|
||||
scored_items = []
|
||||
today_str = get_today_str()
|
||||
|
||||
for item in items:
|
||||
name_val = item.get('name', '')
|
||||
path_val = item.get('path', '')
|
||||
target_str = name_val if name_val else path_val.split('/')[-1]
|
||||
|
||||
try:
|
||||
if is_date_level:
|
||||
current_date = datetime.strptime(target_str, "%Y_%m_%d")
|
||||
else:
|
||||
mod_str = item.get('modified', '')
|
||||
current_date = datetime.fromisoformat(mod_str.replace('Z', '+00:00'))
|
||||
|
||||
diff = abs((today - current_date.replace(tzinfo=None)).total_seconds())
|
||||
scored_items.append((diff, item, target_str))
|
||||
except:
|
||||
continue
|
||||
|
||||
if not scored_items: return None
|
||||
scored_items.sort(key=lambda x: x[0])
|
||||
return scored_items[0]
|
||||
|
||||
|
||||
def process_site(proxy_item):
|
||||
name = proxy_item.get('name', '')
|
||||
# 严格过滤：只处理以 _data 结尾的站点
|
||||
if not name.lower().endswith('_data'):
|
||||
return
|
||||
|
||||
name_upper = name.upper()
|
||||
is_tower_underscore = "TOWER_" in name_upper
|
||||
is_tower_i = "TOWER" in name_upper and not is_tower_underscore
|
||||
|
||||
if not (is_tower_underscore or is_tower_i):
|
||||
return
|
||||
|
||||
print(f"\n--- [正在处理站点: {name}] ---")
|
||||
try:
|
||||
port = proxy_item.get('conf', {}).get('remote_port')
|
||||
if not port: return
|
||||
|
||||
# 动态获取当前站点的 Token
|
||||
token = get_dynamic_token(port)
|
||||
if not token:
|
||||
print(f" 跳过: 无法获取有效的 x-auth")
|
||||
return
|
||||
|
||||
headers = {
|
||||
"Authorization": PRIMARY_AUTH,
|
||||
"x-auth": token,
|
||||
"User-Agent": "Mozilla/5.0"
|
||||
}
|
||||
today_str = get_today_str()
|
||||
|
||||
# Step 1: 进入 data 根目录 (路径根据类型区分大小写)
|
||||
api_root = "/api/resources/Data/" if is_tower_underscore else "/api/resources/data/"
|
||||
res1 = requests.get(f"http://106.75.72.40:{port}{api_root}", headers=headers, timeout=10)
|
||||
|
||||
# Step 2: 寻找并进入日期目录
|
||||
items1 = res1.json().get('items', [])
|
||||
best_date = find_closest_item(items1, is_date_level=True)
|
||||
if not best_date or best_date[2] != today_str:
|
||||
print(f" ⚠ 跳过: 未找到今日 ({today_str}) 的文件夹")
|
||||
return
|
||||
|
||||
date_path = f"{api_root}{best_date[2]}/"
|
||||
res2 = requests.get(f"http://106.75.72.40:{port}{date_path}", headers=headers, timeout=10)
|
||||
|
||||
# Step 3: 寻找日期目录下的最新文件
|
||||
items2 = res2.json().get('items', [])
|
||||
best_file = find_closest_item(items2, is_date_level=False)
|
||||
if not best_file:
|
||||
print(f" ❌ 日期文件夹内无文件")
|
||||
return
|
||||
|
||||
# 获取文件的完整 path
|
||||
file_item = best_file[1]
|
||||
full_path = file_item.get('path')
|
||||
if not full_path:
|
||||
full_path = f"{date_path}{file_item.get('name')}"
|
||||
|
||||
# --- 获取内容层级 (根据站点类型采用不同接口) ---
|
||||
if is_tower_i:
|
||||
# TowerI 模式：使用 /api/raw/{path} 获取二进制流
|
||||
download_url = f"http://106.75.72.40:{port}/api/raw{full_path}"
|
||||
res3 = requests.get(download_url, headers=headers, timeout=20, stream=True)
|
||||
if res3.status_code == 200:
|
||||
save_path = os.path.join(SAVE_DIR, f"{name}_{today_str}.bin")
|
||||
with open(save_path, 'wb') as f:
|
||||
f.write(res3.content)
|
||||
print(f" ✅ 二进制保存成功: {save_path} (大小: {len(res3.content)} 字节)")
|
||||
else:
|
||||
# Tower_ 模式：原来的 JSON content 模式
|
||||
file_api_url = f"http://106.75.72.40:{port}/api/resources{full_path}"
|
||||
res3 = requests.get(file_api_url, headers=headers, timeout=20)
|
||||
try:
|
||||
content_str = res3.json().get('content', '')
|
||||
if content_str:
|
||||
save_path = os.path.join(SAVE_DIR, f"{name}_{today_str}.json")
|
||||
with open(save_path, 'w', encoding='utf-8') as f:
|
||||
f.write(content_str)
|
||||
print(f" ✅ JSON数据保存成功: {save_path}")
|
||||
except:
|
||||
print(f" ❌ TOWER_ 站点格式错误或无内容")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ 站点处理失败: {str(e)}")
|
||||
|
||||
|
||||
def main():
|
||||
print(f"任务启动: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
# 主接口获取站点列表
|
||||
try:
|
||||
# 注意：主控制台可能也需要 token，这里先用硬编码，如果不行也需要为 7500 端口做登录
|
||||
main_headers = {"Authorization": PRIMARY_AUTH, "User-Agent": "Mozilla/5.0"}
|
||||
resp = requests.get(BASE_URL, headers=main_headers, timeout=15)
|
||||
proxies = resp.json().get('proxies', [])
|
||||
for p in proxies:
|
||||
process_site(p)
|
||||
except Exception as e:
|
||||
print(f"❌ 全局错误: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user