Files
api-python/main.py
T
2026-07-02 15:59:08 +08:00

657 lines
24 KiB
Python
Executable File
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from flask import Flask, jsonify, request, redirect, send_file
from datetime import datetime, date
import ipaddress
from zoneinfo import ZoneInfo, available_timezones
from flask_cors import CORS
from lunar_python import Solar
import ip2region.searcher as xdb
import ip2region.util as util
from pathlib import Path
import json
import random
import urllib.error
import urllib.parse
import urllib.request
class LowercasePathMiddleware:
def __init__(self, app, exclude_extensions=None):
self.app = app
self.exclude_extensions = exclude_extensions or set()
def __call__(self, environ, start_response):
path = environ.get("PATH_INFO", "")
if path:
last_segment = path.rsplit('/', 1)[-1]
if '.' not in last_segment:
environ['PATH_INFO'] = path.lower()
return self.app(environ, start_response)
app = Flask(__name__)
CORS(app)
app.json.ensure_ascii = False
app.json.sort_keys = False
app.wsgi_app = LowercasePathMiddleware(app.wsgi_app)
with (Path(__file__).parent / "data" / "luck-data.json").open("r", encoding="utf-8") as f:
LUCK_DATA = json.load(f)
with (Path(__file__).parent / "data" / "poem-300.json").open("r", encoding="utf-8") as f:
POEM_DATA = json.load(f)
with (Path(__file__).parent / "data" / "riddle.json").open("r", encoding="utf-8") as f:
RIDDLE_DATA = json.load(f)
IP2REGION_DB_PATH = str(Path(__file__).parent / "data" / "ip2region_v4.xdb")
IP2REGION_VERSION = util.IPv4
IP2REGION_CONTENT = util.load_content_from_file(IP2REGION_DB_PATH)
IP2REGION_SEARCHER = xdb.new_with_buffer(IP2REGION_VERSION, IP2REGION_CONTENT)
ALL_TIMEZONES = sorted(available_timezones())
LOWER_TIMEZONE_MAP = {tz.lower(): tz for tz in ALL_TIMEZONES}
LUCK_ID_MAP = {item.get("id"): item for item in LUCK_DATA}
LUCK_TITLE_MAP = {}
LUCK_TYPE_MAP = {}
for item in LUCK_DATA:
title_lower = item.get("title", "").lower()
type_lower = item.get("type", "").lower()
if title_lower not in LUCK_TITLE_MAP:
LUCK_TITLE_MAP[title_lower] = []
if type_lower not in LUCK_TYPE_MAP:
LUCK_TYPE_MAP[type_lower] = []
LUCK_TITLE_MAP[title_lower].append(item)
LUCK_TYPE_MAP[type_lower].append(item)
POEM_ID_MAP = {item.get("id"): item for item in POEM_DATA}
POEM_TITLE_MAP = {}
POEM_AUTHOR_MAP = {}
POEM_TYPE_MAP = {}
for item in POEM_DATA:
title_lower = item.get("title", "").lower()
author_lower = item.get("author", "").lower()
type_lower = item.get("type", "").lower()
if title_lower not in POEM_TITLE_MAP:
POEM_TITLE_MAP[title_lower] = []
if author_lower not in POEM_AUTHOR_MAP:
POEM_AUTHOR_MAP[author_lower] = []
if type_lower not in POEM_TYPE_MAP:
POEM_TYPE_MAP[type_lower] = []
POEM_TITLE_MAP[title_lower].append(item)
POEM_AUTHOR_MAP[author_lower].append(item)
POEM_TYPE_MAP[type_lower].append(item)
RIDDLE_RIDDLE_MAP = {}
RIDDLE_ANSWER_MAP = {}
for item in RIDDLE_DATA:
riddle_lower = item.get("riddle", "").lower()
answer_lower = item.get("answer", "").lower()
if riddle_lower not in RIDDLE_RIDDLE_MAP:
RIDDLE_RIDDLE_MAP[riddle_lower] = []
if answer_lower not in RIDDLE_ANSWER_MAP:
RIDDLE_ANSWER_MAP[answer_lower] = []
RIDDLE_RIDDLE_MAP[riddle_lower].append(item)
RIDDLE_ANSWER_MAP[answer_lower].append(item)
@app.route("/")
def index():
return """
<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"></head>
<body><style>table{border-collapse:collapse;width:100%;max-width:500px;margin:20px 0;font-family:system-ui, -apple-system, sans-serif;}
th,td{border:1px solid #e5e7eb;padding:10px 14px;text-align:left;}
th{background-color:#f9fafb;font-weight:600;color:#374151;}
a{color:#2563eb;text-decoration:none;}
a:hover{text-decoration:underline;}</style>
<table><thead><tr><th>接口名称</th><th>API 地址</th><th>说明文档</th></tr></thead><tbody>
<tr><td>音乐接口</td><td><a href="/music/api">/music/api</a></td><td><a href="/music/help">/music/help</a></td></tr>
<tr><td>农历接口</td><td><a href="/lunar/api">/lunar/api</a></td><td><a href="/lunar/help">/lunar/help</a></td></tr>
<tr><td>运势接口</td><td><a href="/luck/api">/luck/api</a></td><td><a href="/luck/help">/luck/help</a></td></tr>
<tr><td>唐诗接口</td><td><a href="/poem/api">/poem/api</a></td><td><a href="/poem/help">/poem/help</a></td></tr>
<tr><td>地址接口</td><td><a href="/ip/api">/ip/api</a></td><td><a href="/ip/help">/ip/help</a></td></tr>
<tr><td>时间接口</td><td><a href="/time/api">/time/api</a></td><td><a href="/time/help">/time/help</a></td></tr>
<tr><td>谜语接口</td><td><a href="/riddle/api">/riddle/api</a></td><td><a href="/riddle/help">/riddle/help</a></td></tr>
<tr><td>期刊接口</td><td><a href="/sci/api">/sci/api</a></td><td><a href="/sci/help">/sci/help</a></td></tr>
<tr><td>下载接口</td><td><a href="/pan/api">/pan/api</a></td><td><a href="/pan/help">/pan/help</a></td></tr>
</tbody></table></body></html>
"""
@app.route("/music/help")
def music_help():
return jsonify({
"路径": "/music/api",
"功能": "获取音乐信息",
"参数": {
"server=netease/tencent/kugou/baidu/kuwo": "必填,音乐平台",
"type=search/song/album/artist/playlist/lrc/url/pic": "必填,操作类型",
"id=xxx": "必填,资源ID",
"author=xxx": "选填,仅lrc/url/pic需要",
},
})
@app.route("/lunar/help")
def lunar_help():
return jsonify({
"路径": "/lunar/api",
"功能": "获取农历信息",
"参数": {
"date=YYYY-MM-DD": "选填,指定日期",
},
})
@app.route("/lunar/api")
def lunar():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"date"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"date"}}参数非法",
"allowed_params": ["date=YYYY-MM-DD"],
})
ds = args_lower.get("date", "").strip()
d = datetime.now(ZoneInfo("Asia/Shanghai")).date()
params = {"date": ds if ds else f"{d}(默认,Asia/Shanghai)"}
if ds:
try:
d = datetime.strptime(ds, "%Y-%m-%d").date()
except ValueError:
resp = {"error": f"date参数[{ds}]不为YYYY-MM-DD格式"}
if params:
resp["params"] = params
return jsonify(resp)
solar = Solar.fromYmd(d.year, d.month, d.day)
lunar = solar.getLunar()
jieqi = lunar.getJieQi() or ""
festivals = lunar.getFestivals()
traditional_festival = festivals[0] if festivals else ""
lunar_month = lunar.getMonthInChinese()
season_map = {"立春": "春", "立夏": "夏", "立秋": "秋", "立冬": "冬"}
latest = None
for j in season_map.keys():
solar_jieqi = Solar.fromYmd(d.year, 1, 1).getLunar().getJieQiTable()[j]
jd = date(solar_jieqi.getYear(), solar_jieqi.getMonth(), solar_jieqi.getDay())
if jd <= d:
if latest is None or jd > latest[0]:
latest = (jd, season_map[j])
resp = {
"nian": lunar.getYearInGanZhi(),
"yue": lunar_month,
"ri": lunar.getDayInChinese(),
"shengxiao": lunar.getYearShengXiao(),
"jijie": latest[1] if latest else "冬",
"jieri": traditional_festival,
"jieqi": jieqi,
}
resp["params"] = params
return jsonify(resp)
@app.route("/luck/help")
def luck_help():
return jsonify({
"路径": "/luck/api",
"功能": "获取运势信息",
"参数": {
"id=xxx": "选填,基于排序查询",
"title=xxx": "选填,基于签名查询",
"type=xxx": "选填,基于类型查询",
},
})
@app.route("/luck/api")
def luck():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"id", "title", "type"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"id", "title", "type"}}参数非法",
"allowed_params": ["id=xxx", "title=xxx", "type=xxx"],
})
luck_id = args_lower.get("id", "").strip()
luck_title = args_lower.get("title", "").strip()
luck_type = args_lower.get("type", "").strip()
params = {"id": luck_id, "title": luck_title, "type": luck_type}
if not luck_id and not luck_title and not luck_type:
resp = random.choice(LUCK_DATA)
return jsonify(resp)
results = LUCK_DATA
if luck_id:
try:
search_id = int(luck_id)
item = LUCK_ID_MAP.get(search_id)
results = [item] if item else []
except ValueError:
resp = {"error": f"id参数[{luck_id}]不为整数"}
resp["params"] = params
return jsonify(resp)
if luck_title:
title_lower = luck_title.lower()
if luck_id:
results = [item for item in results if title_lower in item.get("title", "").lower()]
else:
results = LUCK_TITLE_MAP.get(title_lower, [])
if luck_type:
type_lower = luck_type.lower()
if luck_id or luck_title:
results = [item for item in results if type_lower in item.get("type", "").lower()]
else:
results = LUCK_TYPE_MAP.get(type_lower, [])
if not results:
resp = {"error": "未找到符合条件的数据"}
resp["params"] = params
return jsonify(resp)
return jsonify({"items": results, "params": params})
@app.route("/poem/help")
def poem_help():
return jsonify({
"路径": "/poem/api",
"功能": "获取唐诗信息",
"参数": {
"id=xxx": "选填,基于排序查询",
"title=xxx": "选填,基于标题查询",
"author=xxx": "选填,基于作者查询",
"type=xxx": "选填,基于类型查询",
"full=true/false": "选填,是否返回整首诗(默认为false)",
},
})
@app.route("/poem/api")
def poem():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"id", "title", "author", "type", "full"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"id", "title", "author", "type", "full"}}参数非法",
"allowed_params": ["id=xxx", "title=xxx", "author=xxx", "type=xxx", "full=true/false"],
})
poem_id = args_lower.get("id", "").strip()
poem_title = args_lower.get("title", "").strip()
poem_author = args_lower.get("author", "").strip()
poem_type = args_lower.get("type", "").strip()
full_param = args_lower.get("full", "").strip()
params = {"id": poem_id, "title": poem_title, "author": poem_author, "type": poem_type, "full": full_param if full_param else "false(默认)"}
if full_param.lower() not in {"true", "false", ""}:
resp = {"error": f"full参数[{full_param}]不为true/false"}
resp["params"] = params
return jsonify(resp)
full = full_param.lower() == "true"
if not poem_id and not poem_title and not poem_author and not poem_type:
item = random.choice(POEM_DATA)
contents = item.get("contents", "")
if full:
content = "".join(contents.splitlines())
else:
lines = [line for line in contents.split("\n") if line.strip()]
content = random.choice(lines) if lines else ""
resp = {
"id": item.get("id", ""),
"content": content,
"title": item.get("title", ""),
"author": item.get("author", ""),
"type": item.get("type", ""),
}
resp["params"] = params
return jsonify(resp)
results = POEM_DATA
if poem_id:
try:
search_id = int(poem_id)
item = POEM_ID_MAP.get(search_id)
results = [item] if item else []
except ValueError:
resp = {"error": f"id参数[{poem_id}]不为整数"}
resp["params"] = params
return jsonify(resp)
if poem_title:
title_lower = poem_title.lower()
if poem_id:
results = [item for item in results if title_lower in item.get("title", "").lower()]
else:
results = POEM_TITLE_MAP.get(title_lower, [])
if poem_author:
author_lower = poem_author.lower()
if poem_id or poem_title:
results = [item for item in results if author_lower in item.get("author", "").lower()]
else:
results = POEM_AUTHOR_MAP.get(author_lower, [])
if poem_type:
type_lower = poem_type.lower()
if poem_id or poem_title or poem_author:
results = [item for item in results if type_lower in item.get("type", "").lower()]
else:
results = POEM_TYPE_MAP.get(type_lower, [])
if not results:
resp = {"error": "未找到符合条件的数据"}
resp["params"] = params
return jsonify(resp)
formatted_results = []
for item in results:
contents = item.get("contents", "")
if full:
content = "".join(contents.splitlines())
else:
lines = [line for line in contents.split("\n") if line.strip()]
content = random.choice(lines) if lines else ""
formatted_results.append({
"id": item.get("id", ""),
"content": content,
"title": item.get("title", ""),
"author": item.get("author", ""),
"type": item.get("type", ""),
})
return jsonify({"items": formatted_results, "params": params})
@app.route("/ip/help")
def ip_help():
return jsonify({
"路径": "/ip/api",
"功能": "获取IP信息",
"参数": {
"ip=xxx": "选填,IPv4地址",
},
})
@app.route("/ip/api")
def ip():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"ip"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"ip"}}参数非法",
"allowed_params": ["ip=xxx"],
})
query_ip = args_lower.get("ip", "").strip()
client_ip = request.headers.get("X-Forwarded-For", request.remote_addr).split(",")[0].strip()
target_ip = query_ip or client_ip
params = {"ip": query_ip if query_ip else f"{client_ip}(默认)"}
try:
ip_obj = ipaddress.ip_address(target_ip)
if ip_obj.version != 4:
resp = {"error": f"ip参数[{target_ip}]不为IPv4地址"}
resp["params"] = params
return jsonify(resp)
except ValueError:
resp = {"error": f"ip参数[{target_ip}]不为IPv4地址"}
resp["params"] = params
return jsonify(resp)
try:
region = IP2REGION_SEARCHER.search(target_ip)
if region:
region_parts = region.split("|")
resp = {
"ip": target_ip,
"country": region_parts[0] if len(region_parts) > 0 else "",
"province": region_parts[1] if len(region_parts) > 1 else "",
"city": region_parts[2] if len(region_parts) > 2 else "",
"isp": region_parts[3] if len(region_parts) > 3 else "",
"region": region,
}
resp["params"] = params
return jsonify(resp)
except Exception:
pass
resp = {
"ip": target_ip,
"country": "",
"province": "",
"city": "",
"isp": "",
"region": "",
}
resp["params"] = params
return jsonify(resp)
@app.route("/time/help")
def time_help():
return jsonify({
"路径": "/time/api",
"功能": "获取时间信息",
"参数": {
"tz=xxx": "选填,时区(默认为Shanghai",
},
})
@app.route("/time/api")
def time_api():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"tz"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"tz"}}参数非法",
"allowed_params": ["tz=xxx"],
})
tz = args_lower.get("tz", "").strip()
tz_key = None
params = {"tz": tz if tz else "Shanghai(默认)"}
if tz:
low = tz.lower()
exact = LOWER_TIMEZONE_MAP.get(low)
if exact:
tz_key = exact
elif "/" not in tz:
suffix_matches = [name for name in ALL_TIMEZONES if name.lower().endswith(f"/{low}")]
if len(suffix_matches) == 1:
tz_key = suffix_matches[0]
else:
fuzzy_matches = [name for name in ALL_TIMEZONES if low in name.lower()]
if len(fuzzy_matches) == 1:
tz_key = fuzzy_matches[0]
else:
fuzzy_matches = [name for name in ALL_TIMEZONES if low in name.lower()]
if len(fuzzy_matches) == 1:
tz_key = fuzzy_matches[0]
if tz_key is None:
resp = {"error": f"tz参数[{tz}]不为有效时区名称"}
resp["params"] = params
return jsonify(resp)
else:
tz_key = "Asia/Shanghai"
try:
tzinfo = ZoneInfo(tz_key)
except Exception:
tzinfo = ZoneInfo("Asia/Shanghai")
now = datetime.now(tzinfo)
weekday_iso = now.isoweekday()
week_number = now.isocalendar()[1]
resp = {
"date": now.strftime("%Y-%m-%d"),
"time": now.strftime("%H:%M:%S"),
"timestamp": now.timestamp(),
"tz": tzinfo.key,
"weekday": weekday_iso,
"week": week_number,
}
resp["params"] = params
return jsonify(resp)
@app.route("/riddle/help")
def riddle_help():
return jsonify({
"路径": "/riddle/api",
"功能": "获取谜语信息",
"参数": {
"riddle=xxx": "选填,基于谜面查询",
"answer=xxx": "选填,基于谜底查询",
},
})
@app.route("/riddle/api")
def riddle():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"riddle", "answer"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"riddle", "answer"}}参数非法",
"allowed_params": ["riddle=xxx", "answer=xxx"],
})
query_riddle = args_lower.get("riddle", "").strip()
query_answer = args_lower.get("answer", "").strip()
params = {"riddle": query_riddle, "answer": query_answer}
if not query_riddle and not query_answer:
item = random.choice(RIDDLE_DATA)
resp = {
"riddle": item.get("riddle", ""),
"answer": item.get("answer", ""),
}
resp["params"] = params
return jsonify(resp)
results = RIDDLE_DATA
if query_riddle:
riddle_lower = query_riddle.lower()
results = RIDDLE_RIDDLE_MAP.get(riddle_lower, [])
if query_answer:
answer_lower = query_answer.lower()
if query_riddle:
results = [item for item in results if answer_lower in item.get("answer", "").lower()]
else:
results = RIDDLE_ANSWER_MAP.get(answer_lower, [])
if not results:
resp = {"error": "未找到符合条件的数据"}
resp["params"] = params
return jsonify(resp)
items = [{
"riddle": item.get("riddle", ""),
"answer": item.get("answer", ""),
} for item in results]
return jsonify({"items": items, "params": params})
@app.route("/sci/help")
def sci_help():
return jsonify({
"路径": "/sci/api",
"功能": "获取期刊信息",
"参数": {
"nane=xxx": "必填,期刊名称(可缩写)",
},
})
@app.route("/sci/api")
def sci():
args_lower = {k.lower(): v for k, v in request.args.items()}
if set(args_lower.keys()) - {"nane", "name"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"nane", "name"}}参数非法",
"allowed_params": ["nane=xxx"],
})
publication_name = args_lower.get("nane", "").strip() or args_lower.get("name", "").strip()
params = {"nane": publication_name}
if not publication_name:
resp = {"error": "nane参数不能为空"}
resp["params"] = params
return jsonify(resp)
query = urllib.parse.urlencode({
"secretKey": "0361aa1c9a8b4a30889ab6b02eae902c",
"publicationName": publication_name,
})
url = f"https://www.easyscholar.cc/open/getPublicationRank?{query}"
try:
request_obj = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(request_obj, timeout=10) as response:
raw_data = response.read().decode("utf-8")
except urllib.error.HTTPError as error:
resp = {"error": f"期刊接口请求失败: HTTP {error.code}"}
resp["params"] = params
return jsonify(resp)
except urllib.error.URLError as error:
resp = {"error": f"期刊接口请求失败: {error.reason}"}
resp["params"] = params
return jsonify(resp)
try:
result = json.loads(raw_data)
except json.JSONDecodeError:
resp = {"error": "期刊接口返回内容不是有效JSON"}
resp["params"] = params
return jsonify(resp)
if result.get("code") != 200:
resp = {"error": result.get("msg", "期刊接口返回异常")}
resp["params"] = params
return jsonify(resp)
data = result.get("data") or {}
official_rank = data.get("officialRank") or {}
all_rank = official_rank.get("all") or {}
resp = {
"CCF等级": all_rank.get("ccf", ""),
"影响因子": all_rank.get("sciif", ""),
"JCR分区": all_rank.get("sci", ""),
"中科院分区": all_rank.get("sciUp", ""),
"TOP期刊": all_rank.get("sciUpTop", ""),
}
resp["params"] = params
return jsonify(resp)
@app.route("/pan/help")
def pan_help():
return jsonify({
"路径": "/pan/api",
"功能": "下载文件",
"参数": {
"name=xxx": "必填,文件名",
"download=true/false": "下载/预览",
},
"列表": ["anian-music.js"]
})
@app.route("/pan/api")
def pan_file_download():
args_lower = {k.lower(): v for k, v in request.args.items()}
allowed_keys = {"name", "download"}
extra_params = set(args_lower.keys()) - allowed_keys
if extra_params:
return jsonify({
"error": f"{extra_params}参数非法",
"allowed_params": ["name=xxx", "download=true/false"],
})
file_name = args_lower.get("name", "").strip()
download_param = args_lower.get("download", "").strip()
if download_param and download_param.lower() not in {"true", "false"}:
resp = {
"error": f"download参数[{download_param}]不为true/false",
"params": {"name": file_name, "download": download_param}
}
return jsonify(resp)
force_download = download_param.lower() == "true"
params = {"name": file_name, "download": download_param if download_param else "false(默认)"}
if not file_name:
resp = {
"error": "name参数不能为空",
"params": params
}
return jsonify(resp)
pan_dir = Path(__file__).parent / "data" / "pan"
target_file = pan_dir / file_name
real_pan = pan_dir.resolve()
real_target = target_file.resolve()
if not str(real_target).startswith(str(real_pan)):
resp = {
"error": "文件名非法",
"params": params
}
return jsonify(resp)
if not target_file.is_file():
resp = {
"error": f"文件 {file_name} 不存在",
"help": "请前往/pan/help查看文件列表",
"params": params
}
return jsonify(resp)
# 在线预览/播放核心:as_attachment=False
return send_file(
target_file,
as_attachment=force_download,
download_name=file_name
)
@app.errorhandler(404)
def not_found(_error):
return redirect("/"), 302
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)