import os, json, random, socket, logging, ipaddress
from pathlib import Path
from datetime import date, datetime
from zoneinfo import ZoneInfo, available_timezones
from urllib import error, parse, request
from flask import Flask, jsonify, request, redirect, send_file, Response
from flask_cors import CORS
from lunar_python import Solar
from ip2region import searcher as xdb, util
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)
logging.getLogger("werkzeug").setLevel(logging.WARNING)
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("/favicon.ico")
def favicon():
logo_file = os.path.join(app.root_path, "data/pan", "zhaoxi-logo.png")
return send_file(logo_file, mimetype="image/png")
@app.route("/")
def index():
return """
"""
@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信息",
"参数": {
"name=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()) - {"name"}:
return jsonify({
"error": f"{set(args_lower.keys()) - {"name"}}参数非法",
"allowed_params": ["name=xxx"],
})
query_name = args_lower.get("name", "").strip()
client_ip = request.headers.get("X-Forwarded-For", request.remote_addr).split(",")[0].strip()
target = query_name or client_ip
params = {"name": query_name if query_name else f"{client_ip}(默认)"}
try:
ip_obj = ipaddress.ip_address(target)
if ip_obj.version != 4:
resp = {"error": f"name参数[{target}]不为IPv4地址或可解析为IPv4的域名"}
resp["params"] = params
return jsonify(resp)
target_ip = target
except ValueError:
try:
addr_infos = socket.getaddrinfo(target, None, socket.AF_INET)
except socket.gaierror:
resp = {"error": f"name参数[{target}]不为IPv4地址或可解析为IPv4的域名"}
resp["params"] = params
return jsonify(resp)
resolved_ips = sorted({info[4][0] for info in addr_infos})
if not resolved_ips:
resp = {"error": f"name参数[{target}]未解析到IPv4地址"}
resp["params"] = params
return jsonify(resp)
target_ip = resolved_ips[0]
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("/agent/help")
def agent_help():
return jsonify({
"路径": "/agent/api",
"功能": "访问目标网址并返回内容",
"参数": {
"url=xxx": "必填，目标网址（仅支持http/https）",
"ref=xxx": "选填，Referer（默认为空）",
"pwd=xxx": "选填，访问白名单以外网址时必填",
},
"白名单": ["anian.net", "*.anian.net"],
})
@app.route("/agent/api")
def agent():
agent_password = "anian-agent-api-pwd"
whitelist_domains = {"anian.net"}
args_lower = {k.lower(): v for k, v in request.args.items()}
allowed_keys = {"url", "ref", "pwd", "password"}
extra_params = set(args_lower.keys()) - allowed_keys
if extra_params:
return jsonify({
"error": f"{extra_params}参数非法",
"allowed_params": ["url=xxx", "ref=xxx", "pwd=xxx"],
})
target_url = args_lower.get("url", "").strip()
referer = args_lower.get("ref", "").strip()
password = args_lower.get("pwd", "").strip() or args_lower.get("password", "").strip()
params = {"url": target_url, "ref": referer, "pwd": password}
if not target_url:
resp = {"error": "url参数不能为空"}
resp["params"] = params
return jsonify(resp)
parsed_url = urllib.parse.urlparse(target_url)
if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
resp = {"error": f"url参数[{target_url}]不为有效http/https网址"}
resp["params"] = params
return jsonify(resp)
hostname = (parsed_url.hostname or "").rstrip(".").lower()
try:
ipaddress.ip_address(hostname)
resp = {"error": "url参数只允许填写域名，不允许IP地址"}
resp["params"] = params
return jsonify(resp)
except ValueError:
pass
is_whitelisted = any(
hostname == domain or hostname.endswith(f".{domain}")
for domain in whitelist_domains
)
if not is_whitelisted and password != agent_password:
resp = {"error": "非白名单URL需要正确密码"}
resp["params"] = params
return jsonify(resp)
try:
addr_infos = socket.getaddrinfo(hostname, None)
except socket.gaierror as error:
resp = {"error": f"域名解析失败: {error.strerror or error}"}
resp["params"] = params
return jsonify(resp)
resolved_ips = sorted({info[4][0] for info in addr_infos})
for resolved_ip in resolved_ips:
ip_obj = ipaddress.ip_address(resolved_ip)
if not ip_obj.is_global:
resp = {"error": f"域名解析到非公网IP: {resolved_ip}"}
resp["params"] = params
return jsonify(resp)
headers = {
"User-Agent": "Mozilla/5.0",
"Referer": referer,
}
try:
request_obj = urllib.request.Request(target_url, headers=headers)
with urllib.request.urlopen(request_obj, timeout=15) as response:
body = response.read()
content_type = response.headers.get("Content-Type", "application/octet-stream")
return Response(body, status=response.getcode(), content_type=content_type)
except urllib.error.HTTPError as error:
body = error.read()
content_type = error.headers.get("Content-Type", "text/plain; charset=utf-8")
return Response(body, status=error.code, content_type=content_type)
except urllib.error.URLError as error:
resp = {"error": f"agent请求失败: {error.reason}"}
resp["params"] = params
return jsonify(resp)
except ValueError as error:
resp = {"error": f"agent请求失败: {error}"}
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", "zhaoxi-logo.png"]
})
@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)
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)