新增速率限制；增强兼容性

README.md

@@ -1,6 +1,17 @@
 # API-Python
 
-> 基于Python的本地Web API项目
+> 基于Python3.9+的Web API项目
+
+#### v260728
+1. 优化（修复）import
+2. 修改README
+3. 新增IP请求速率限制（默认30/min）
+4. 增强兼容性
+
+#### v260716
+1. 新增代理接口（包含白名单及密码限制）
+2. 地址接口新增域名支持
+3. 取消音乐接口
 
 #### v260702
 1. 新增下载接口，同时提供web播放器anian-music.js下载
@@ -10,3 +21,5 @@
 
 #### v260512
 1. 新增农历、运势、唐诗、地址、时间、谜语等接口及对应数据
+
+

main.py

@@ -1,17 +1,14 @@
-from flask import Flask, jsonify, request, redirect, send_file
+import os, json, random, socket, logging, ipaddress, urllib.error, urllib.parse, urllib.request, threading
-from datetime import datetime, date
+from collections import defaultdict, deque
-import ipaddress
+from time import monotonic
+from pathlib import Path
+from datetime import date, datetime
 from zoneinfo import ZoneInfo, available_timezones
+
+from flask import Flask, jsonify, request, redirect, send_file, Response
 from flask_cors import CORS
 from lunar_python import Solar
-import ip2region.searcher as xdb
+from ip2region import searcher as xdb, util
-import ip2region.util as util
-from pathlib import Path
-import json
-import random
-import urllib.error
-import urllib.parse
-import urllib.request
 
 
 class LowercasePathMiddleware:
@@ -32,6 +29,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:
@@ -94,6 +92,39 @@ for item in RIDDLE_DATA:
     RIDDLE_RIDDLE_MAP[riddle_lower].append(item)
     RIDDLE_ANSWER_MAP[answer_lower].append(item)
 
+RATE_LIMIT = 30
+RATE_WINDOW_SECONDS = 60
+_request_times = defaultdict(deque)
+_rate_limit_lock = threading.Lock()
+
+
+@app.before_request
+def enforce_rate_limit():
+    now = monotonic()
+    forwarded_for = request.headers.get("X-Forwarded-For", "")
+    client_ip = forwarded_for.split(",", 1)[0].strip() or request.remote_addr or "unknown"
+    with _rate_limit_lock:
+        timestamps = _request_times[client_ip]
+        cutoff = now - RATE_WINDOW_SECONDS
+        while timestamps and timestamps[0] <= cutoff:
+            timestamps.popleft()
+        if len(timestamps) >= RATE_LIMIT:
+            retry_after = max(1, int(timestamps[0] + RATE_WINDOW_SECONDS - now + 0.999))
+            response = jsonify({
+                "error": "请求过于频繁，请稍后再试",
+                "limit": f"{RATE_LIMIT}/min",
+                "retry_after": retry_after,
+            })
+            response.status_code = 429
+            response.headers["Retry-After"] = str(retry_after)
+            return response
+        timestamps.append(now)
+
+
+@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():
@@ -105,7 +136,6 @@ 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>
@@ -113,24 +143,12 @@ a:hover{text-decoration:underline;}</style>
 <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="/agent/api">/agent/api</a></td><td><a href="/agent/help">/agent/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():
@@ -147,7 +165,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"date"}),
             "allowed_params": ["date=YYYY-MM-DD"],
         })
     ds = args_lower.get("date", "").strip()
@@ -205,7 +223,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"id", "title", "type"}),
             "allowed_params": ["id=xxx", "title=xxx", "type=xxx"],
         })
     luck_id = args_lower.get("id", "").strip()
@@ -263,7 +281,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(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()
@@ -350,32 +368,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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"name"}),
-            "allowed_params": ["ip=xxx"],
+            "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
+    target = query_name or client_ip
-    params = {"ip": query_ip if query_ip else f"{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地址"}
+        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:
@@ -419,7 +447,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"tz"}),
             "allowed_params": ["tz=xxx"],
         })
     tz = args_lower.get("tz", "").strip()
@@ -484,7 +512,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"riddle", "answer"}),
             "allowed_params": ["riddle=xxx", "answer=xxx"],
         })
     query_riddle = args_lower.get("riddle", "").strip()
@@ -534,7 +562,7 @@ 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"}}参数非法",
+            "error": "{}参数非法".format(set(args_lower.keys()) - {"nane", "name"}),
             "allowed_params": ["nane=xxx"],
         })
     publication_name = args_lower.get("nane", "").strip() or args_lower.get("name", "").strip()
@@ -584,6 +612,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": "{}参数非法".format(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 +712,7 @@ def pan_help():
             "name=xxx": "必填，文件名",
             "download=true/false": "下载/预览",
             },
-        "列表": ["anian-music.js"]
+        "列表": ["anian-music.js", "zhaoxi-logo.png"]
     })
 
 @app.route("/pan/api")
@@ -603,7 +722,7 @@ def pan_file_download():
     extra_params = set(args_lower.keys()) - allowed_keys
     if extra_params:
         return jsonify({
-            "error": f"{extra_params}参数非法",
+            "error": "{}参数非法".format(extra_params),
             "allowed_params": ["name=xxx", "download=true/false"],
         })
     file_name = args_lower.get("name", "").strip()
@@ -639,8 +758,6 @@ def pan_file_download():
             "params": params
         }
         return jsonify(resp)
-
-    # 在线预览/播放核心：as_attachment=False
     return send_file(
         target_file,
         as_attachment=force_download,

requirements.txt

@@ -2,3 +2,4 @@ flask
 flask-cors
 lunar-python
 py-ip2region
+tzdata