diff --git a/README.md b/README.md index 975e9e4..ae1e3f8 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,12 @@ -# API-Python +# Python-API > 基于Python的本地Web API项目 +#### v260716 +1. 新增代理接口（包含白名单及密码限制） +2. 地址接口新增域名支持 +3. 取消音乐接口 + #### v260702 1. 新增下载接口，同时提供web播放器anian-music.js下载 diff --git a/data/pan/zhaoxi-logo.png b/data/pan/zhaoxi-logo.png new file mode 100644 index 0000000..7ac8dd6 Binary files /dev/null and b/data/pan/zhaoxi-logo.png differ diff --git a/main.py b/main.py index fc6023d..f5263ce 100755 --- a/main.py +++ b/main.py @@ -1,4 +1,4 @@ -from flask import Flask, jsonify, request, redirect, send_file +from flask import Flask, jsonify, request, redirect, send_file, Response from datetime import datetime, date import ipaddress from zoneinfo import ZoneInfo, available_timezones @@ -12,6 +12,9 @@ import random import urllib.error import urllib.parse import urllib.request +import logging +import socket +import os class LowercasePathMiddleware: @@ -32,6 +35,7 @@ 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: @@ -95,6 +99,11 @@ for item in RIDDLE_DATA: 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 """ @@ -105,7 +114,6 @@ th{background-color:#f9fafb;font-weight:600;color:#374151;} a{color:#2563eb;text-decoration:none;} a:hover{text-decoration:underline;} - @@ -113,24 +121,12 @@ a:hover{text-decoration:underline;} +
接口名称API 地址说明文档
音乐接口/music/api/music/help
农历接口/lunar/api/lunar/help
运势接口/luck/api/luck/help
唐诗接口/poem/api/poem/help
时间接口/time/api/time/help
谜语接口/riddle/api/riddle/help
期刊接口/sci/api/sci/help
代理接口/agent/api/agent/help
下载接口/pan/api/pan/help
""" -@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(): @@ -350,32 +346,42 @@ def ip_help(): "路径": "/ip/api", "功能": "获取IP信息", "参数": { - "ip=xxx": "选填，IPv4地址", + "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()) - {"ip"}: + if set(args_lower.keys()) - {"name"}: return jsonify({ - "error": f"{set(args_lower.keys()) - {"ip"}}参数非法", - "allowed_params": ["ip=xxx"], + "error": f"{set(args_lower.keys()) - {"name"}}参数非法", + "allowed_params": ["name=xxx"], }) - query_ip = args_lower.get("ip", "").strip() + query_name = args_lower.get("name", "").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}(默认)"} + target = query_name or client_ip + params = {"name": query_name if query_name else f"{client_ip}(默认)"} try: - ip_obj = ipaddress.ip_address(target_ip) + ip_obj = ipaddress.ip_address(target) if ip_obj.version != 4: - resp = {"error": f"ip参数[{target_ip}]不为IPv4地址"} + resp = {"error": f"name参数[{target}]不为IPv4地址或可解析为IPv4的域名"} resp["params"] = params return jsonify(resp) + target_ip = target except ValueError: - resp = {"error": f"ip参数[{target_ip}]不为IPv4地址"} - resp["params"] = params - return jsonify(resp) + 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: @@ -584,6 +590,97 @@ def sci(): 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({ @@ -593,7 +690,7 @@ def pan_help(): "name=xxx": "必填，文件名", "download=true/false": "下载/预览", }, - "列表": ["anian-music.js"] + "列表": ["anian-music.js", "zhaoxi-logo.png"] }) @app.route("/pan/api") @@ -639,8 +736,6 @@ def pan_file_download(): "params": params } return jsonify(resp) - - # 在线预览/播放核心：as_attachment=False return send_file( target_file, as_attachment=force_download, @@ -654,4 +749,4 @@ def not_found(_error): if __name__ == "__main__": - app.run(host="0.0.0.0", port=5000, debug=True) \ No newline at end of file + app.run(host="0.0.0.0", port=5000, debug=True)