from flask import Flask, jsonify, request, redirect, send_file, Response
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
import logging
import socket
import os
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)