96 lines
4.0 KiB
Python
96 lines
4.0 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""每日管线健康检查 — 验证所有依赖是否正常，防断线"""
|
||
|
|
import sys, os, io, urllib.request, json, socket
|
||
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
from config import (TELEGRAPH_TOKEN, GITHUB_TOKEN, TG_BOT_TOKEN, DISCORD_WEBHOOK,
|
||
|
|
DEEPSEEK_KEY, INDEXNOW_KEY, BLOG_URL, HTTP_PROXY)
|
||
|
|
|
||
|
|
CHECKS = []
|
||
|
|
|
||
|
|
def check(name, fn):
|
||
|
|
try:
|
||
|
|
ok, msg = fn()
|
||
|
|
icon = "PASS" if ok else "FAIL"
|
||
|
|
print(f" [{icon}] {name}: {msg}")
|
||
|
|
return ok
|
||
|
|
except Exception as e:
|
||
|
|
print(f" [FAIL] {name}: {str(e)[:100]}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def check_telegraph():
|
||
|
|
if not TELEGRAPH_TOKEN:
|
||
|
|
return False, "TELEGRAPH_TOKEN is empty"
|
||
|
|
data = json.dumps({'access_token': TELEGRAPH_TOKEN, 'title': 'test', 'content': [{'tag':'p','children':['test']}], 'return_content': False}).encode()
|
||
|
|
req = urllib.request.Request('https://api.telegra.ph/createAccount', data=data,
|
||
|
|
headers={'Content-Type': 'application/json'}, method='POST')
|
||
|
|
r = urllib.request.urlopen(req, timeout=15)
|
||
|
|
result = json.loads(r.read())
|
||
|
|
return result.get('ok', False), result.get('result', {}).get('short_name', '?')
|
||
|
|
|
||
|
|
def check_github():
|
||
|
|
if not GITHUB_TOKEN:
|
||
|
|
return False, "GITHUB_TOKEN is empty"
|
||
|
|
req = urllib.request.Request('https://api.github.com/user',
|
||
|
|
headers={'Authorization': f'Bearer {GITHUB_TOKEN}', 'User-Agent': 'GFIL-HealthCheck'})
|
||
|
|
r = urllib.request.urlopen(req, timeout=15)
|
||
|
|
data = json.loads(r.read())
|
||
|
|
return data.get('login') is not None, f"User: {data.get('login','?')}"
|
||
|
|
|
||
|
|
def check_indexnow():
|
||
|
|
if not INDEXNOW_KEY:
|
||
|
|
return False, "INDEXNOW_KEY is empty"
|
||
|
|
payload = json.dumps({'host': 'blog.quant-view.xyz', 'key': INDEXNOW_KEY, 'urlList': [f'{BLOG_URL}/']}).encode()
|
||
|
|
req = urllib.request.Request('https://api.indexnow.org/indexnow', data=payload,
|
||
|
|
headers={'Content-Type': 'application/json'})
|
||
|
|
r = urllib.request.urlopen(req, timeout=15)
|
||
|
|
return r.status in (200, 202), f"HTTP {r.status}"
|
||
|
|
|
||
|
|
def check_proxy():
|
||
|
|
if not HTTP_PROXY:
|
||
|
|
return True, "No proxy configured (direct connection)"
|
||
|
|
h, p = HTTP_PROXY.split(':')
|
||
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
|
|
s.settimeout(5)
|
||
|
|
try:
|
||
|
|
s.connect((h, int(p)))
|
||
|
|
s.close()
|
||
|
|
return True, f"Proxy {HTTP_PROXY} alive"
|
||
|
|
except:
|
||
|
|
return False, f"Proxy {HTTP_PROXY} DOWN"
|
||
|
|
|
||
|
|
def check_site():
|
||
|
|
req = urllib.request.Request(f'{BLOG_URL}/',
|
||
|
|
headers={'User-Agent': 'Mozilla/5.0 GFIL-HealthCheck'})
|
||
|
|
r = urllib.request.urlopen(req, timeout=15)
|
||
|
|
return r.status == 200, f"HTTP {r.status}"
|
||
|
|
|
||
|
|
def check_gsc_sa():
|
||
|
|
sa = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'gothic-venture-498218-u0-15afe4efe6f3.json')
|
||
|
|
if not os.path.exists(sa):
|
||
|
|
return False, "Service account JSON not found"
|
||
|
|
try:
|
||
|
|
json.load(open(sa))
|
||
|
|
return True, "Service account JSON valid"
|
||
|
|
except:
|
||
|
|
return False, "Invalid JSON"
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
print(f'=== GFIL Pipeline Health Check ===')
|
||
|
|
print(f'Time: {__import__("datetime").datetime.now().isoformat()}')
|
||
|
|
print(f'Site: {BLOG_URL}\n')
|
||
|
|
|
||
|
|
all_ok = True
|
||
|
|
all_ok &= check("Telegraph API", check_telegraph)
|
||
|
|
all_ok &= check("GitHub API", check_github)
|
||
|
|
all_ok &= check("IndexNow API", check_indexnow)
|
||
|
|
all_ok &= check("Proxy (7890)", check_proxy)
|
||
|
|
all_ok &= check("Site Alive", check_site)
|
||
|
|
all_ok &= check("GSC Service Acct", check_gsc_sa)
|
||
|
|
all_ok &= check("TG Bot Token", lambda: (bool(TG_BOT_TOKEN), f"Token: {'set' if TG_BOT_TOKEN else 'MISSING'}"))
|
||
|
|
all_ok &= check("Discord Webhook", lambda: (bool(DISCORD_WEBHOOK), f"Webhook: {'set' if DISCORD_WEBHOOK else 'MISSING'}"))
|
||
|
|
all_ok &= check("DeepSeek Key", lambda: (bool(DEEPSEEK_KEY), f"Key: {'set' if DEEPSEEK_KEY else 'MISSING'}"))
|
||
|
|
|
||
|
|
print(f'\n{"ALL CHECKS PASSED" if all_ok else "SOME CHECKS FAILED — CHECK ABOVE"}')
|