|
|
from flask import Flask, request, jsonify |
|
|
import re |
|
|
import requests |
|
|
import os |
|
|
from functools import wraps |
|
|
|
|
|
app = Flask(__name__) |
|
|
|
|
|
|
|
|
API_KEY = os.environ.get("XHS_API_KEY", "default_secret_key") |
|
|
|
|
|
|
|
|
def require_api_key(f): |
|
|
@wraps(f) |
|
|
def decorated_function(*args, **kwargs): |
|
|
|
|
|
api_key = request.headers.get('x-api-key') |
|
|
|
|
|
|
|
|
if api_key and api_key == API_KEY: |
|
|
return f(*args, **kwargs) |
|
|
else: |
|
|
return jsonify({"status": "error", "message": "无效的API密钥"}), 401 |
|
|
|
|
|
return decorated_function |
|
|
|
|
|
def extract_url(text): |
|
|
"""从文本中提取小红书链接（短链接或完整链接）""" |
|
|
|
|
|
short_pattern = r'http://xhslink\.com/\S+' |
|
|
|
|
|
full_pattern = r'https://www\.xiaohongshu\.com/\S+' |
|
|
|
|
|
|
|
|
short_matches = re.findall(short_pattern, text) |
|
|
short_urls = [match.rstrip('，').rstrip(',').rstrip('。').rstrip('.') for match in short_matches] |
|
|
|
|
|
|
|
|
full_matches = re.findall(full_pattern, text) |
|
|
full_urls = [match.rstrip('，').rstrip(',').rstrip('。').rstrip('.') for match in full_matches] |
|
|
|
|
|
|
|
|
result = [] |
|
|
for url in short_urls: |
|
|
result.append(("short", url)) |
|
|
for url in full_urls: |
|
|
result.append(("full", url)) |
|
|
|
|
|
return result |
|
|
|
|
|
def get_redirect_url(short_url): |
|
|
"""获取短链接重定向后的真实链接""" |
|
|
try: |
|
|
headers = { |
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", |
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8", |
|
|
"Accept-Language": "en-US,en;q=0.9", |
|
|
"Connection": "keep-alive", |
|
|
} |
|
|
response = requests.get(short_url, headers=headers, allow_redirects=False) |
|
|
if response.status_code in [301, 302, 307]: |
|
|
return response.headers['Location'] |
|
|
else: |
|
|
return f"错误: 无法获取重定向链接，状态码: {response.status_code}" |
|
|
except Exception as e: |
|
|
return f"错误: 请求失败 - {str(e)}" |
|
|
|
|
|
def convert_links(text): |
|
|
"""转换文本中的所有小红书链接""" |
|
|
urls = extract_url(text) |
|
|
if not urls: |
|
|
return {"status": "error", "message": "未找到小红书链接", "results": []} |
|
|
|
|
|
results = [] |
|
|
for url_type, url in urls: |
|
|
if url_type == "short": |
|
|
real_url = get_redirect_url(url) |
|
|
results.append({ |
|
|
"type": "short", |
|
|
"short_url": url, |
|
|
"real_url": real_url |
|
|
}) |
|
|
else: |
|
|
results.append({ |
|
|
"type": "full", |
|
|
"url": url |
|
|
}) |
|
|
|
|
|
return {"status": "success", "message": "链接转换成功", "results": results} |
|
|
|
|
|
@app.route('/api/convert', methods=['POST']) |
|
|
@require_api_key |
|
|
def api_convert(): |
|
|
"""API端点，接收POST请求并转换链接""" |
|
|
if not request.is_json: |
|
|
return jsonify({"status": "error", "message": "请求必须是JSON格式"}), 400 |
|
|
|
|
|
data = request.get_json() |
|
|
|
|
|
if 'text' not in data: |
|
|
return jsonify({"status": "error", "message": "请求中必须包含'text'字段"}), 400 |
|
|
|
|
|
text = data['text'] |
|
|
result = convert_links(text) |
|
|
|
|
|
return jsonify(result) |
|
|
|
|
|
@app.route('/api/health', methods=['GET']) |
|
|
@require_api_key |
|
|
def health_check(): |
|
|
"""健康检查端点""" |
|
|
return jsonify({"status": "ok", "message": "服务正常运行"}) |
|
|
|
|
|
@app.route('/', methods=['GET']) |
|
|
def home(): |
|
|
"""主页，显示简单的使用说明""" |
|
|
return """ |
|
|
<html> |
|
|
<head> |
|
|
<title>小红书链接转换API</title> |
|
|
<style> |
|
|
body { font-family: Arial, sans-serif; margin: 40px; line-height: 1.6; } |
|
|
h1 { color: #D81E06; } |
|
|
code { background: #f4f4f4; padding: 2px 5px; border-radius: 3px; } |
|
|
pre { background: #f4f4f4; padding: 15px; border-radius: 5px; overflow-x: auto; } |
|
|
</style> |
|
|
</head> |
|
|
<body> |
|
|
<h1>小红书链接转换API</h1> |
|
|
<p>这是一个将小红书分享链接转换为真实可访问链接的API服务。</p> |
|
|
|
|
|
<h2>API使用说明</h2> |
|
|
<h3>认证</h3> |
|
|
<p>所有API请求都需要在请求头中包含有效的API密钥：</p> |
|
|
<pre> |
|
|
x-api-key: [您的API密钥] |
|
|
</pre> |
|
|
<p>API密钥可以通过环境变量 <code>XHS_API_KEY</code> 设置</p> |
|
|
|
|
|
<h3>转换链接</h3> |
|
|
<p><strong>端点:</strong> <code>/api/convert</code></p> |
|
|
<p><strong>方法:</strong> POST</p> |
|
|
<p><strong>请求体:</strong></p> |
|
|
<pre> |
|
|
{ |
|
|
"text": "包含小红书链接的文本" |
|
|
} |
|
|
</pre> |
|
|
<p><strong>示例响应:</strong></p> |
|
|
<pre> |
|
|
{ |
|
|
"status": "success", |
|
|
"message": "链接转换成功", |
|
|
"results": [ |
|
|
{ |
|
|
"type": "short", |
|
|
"short_url": "http://xhslink.com/xxxx", |
|
|
"real_url": "https://www.xiaohongshu.com/discovery/item/..." |
|
|
} |
|
|
] |
|
|
} |
|
|
</pre> |
|
|
|
|
|
<h3>健康检查</h3> |
|
|
<p><strong>端点:</strong> <code>/api/health</code></p> |
|
|
<p><strong>方法:</strong> GET</p> |
|
|
<p><strong>示例响应:</strong></p> |
|
|
<pre> |
|
|
{ |
|
|
"status": "ok", |
|
|
"message": "服务正常运行" |
|
|
} |
|
|
</pre> |
|
|
</body> |
|
|
</html> |
|
|
""" |
|
|
|
|
|
if __name__ == '__main__': |
|
|
|
|
|
app.run(host='0.0.0.0', port=7860, debug=False, threaded=True) |
