File size: 9,128 Bytes
b458f3d 67a97b3 b458f3d f2e524e c50a4e1 f2e524e c50a4e1 f2e524e a09c50d 67a97b3 a09c50d 67a97b3 3da7fc2 67a97b3 3da7fc2 67a97b3 b458f3d 6d07192 b458f3d 6d07192 b458f3d 492f326 b458f3d 6d07192 f730cd8 6d07192 b458f3d 6d07192 b458f3d 6d07192 b458f3d 6d07192 b458f3d 6d07192 b458f3d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 | from flask import Flask, jsonify, request
from flask_cors import CORS
import json
import os
import random
from src.crawler import Crawler
from src.analyzer import ImageAnalyzer
from src.monitor import SystemMonitor
app = Flask(__name__)
CORS(app)
REPORT_FILE = 'seo_report.json'
# --- Keep Alive Mechanism ---
import threading
import time
import requests
class KeepAlive(threading.Thread):
def __init__(self, interval=60, target_url="http://127.0.0.1:7860/health"):
super().__init__()
self.interval = interval
self.target_url = target_url
self.daemon = True # Stop when main thread stops
self.running = True
def run(self):
print("KeepAlive System Started")
while self.running:
try:
# 1. Log Heartbeat (keeps logs active)
print(f"[Heartbeat] System Active - {time.ctime()}")
# 2. Self-Ping to keep connection warnings away (Wait for server to start first)
time.sleep(self.interval)
# Try to self-ping if server is likely up (after 10s)
try:
requests.get(self.target_url, timeout=5)
except:
pass # Ignore connection errors during startup/shutdown
except Exception as e:
print(f"[KeepAlive] Error: {e}")
time.sleep(60)
class ScheduledCrawler(threading.Thread):
def __init__(self, interval=14400): # 14400 seconds = 4 Hours
super().__init__()
self.interval = interval
self.daemon = True
self.running = True
def run(self):
print(f"[Scheduler] Auto-Crawler initialized. Schedule: Every {self.interval/3600} hours.")
# Initial delay to let server start
time.sleep(60)
while self.running:
try:
# 1. Identify Target
# User configures this via Environment Variable
target_domain = os.environ.get("AUTO_CRAWL_TARGET")
if target_domain:
print(f"\n[Scheduler] 🕒 Triggering scheduled crawl for: {target_domain}")
# 2. Run Crawl (Reuse logic via internal call or simulating request)
# We instantiate the classes directly to avoid network overhead
crawler = Crawler()
site_data, _, _ = crawler.crawl_domain(target_domain, max_pages=50) # Limit to 50 for auto-runs
if site_data:
analyzer = ImageAnalyzer()
results = analyzer.analyze_site(site_data)
# Save Report
with open(REPORT_FILE, 'w', encoding='utf-8') as f:
json.dump(results, f, indent=4)
print(f"[Scheduler] ✅ Crawl finished for {target_domain}. Report saved.")
else:
print(f"[Scheduler] ⚠️ Crawl returned no data.")
else:
print("[Scheduler] ℹ️ waiting... (Set 'AUTO_CRAWL_TARGET' Env Var to enable auto-crawling)")
# 3. Wait for next interval
time.sleep(self.interval)
except Exception as e:
print(f"[Scheduler] Error: {e}")
time.sleep(60)
# Start KeepAlive & Scheduler
pinger = KeepAlive(interval=300)
pinger.start()
scheduler = ScheduledCrawler(interval=14400) # 4 Hours
scheduler.start()
# ----------------------------
@app.route('/')
def home():
return "Antigravity API is Running. Use /api/status for system info."
@app.route('/health')
def health_check():
return jsonify({"status": "alive"}), 200
@app.route('/api/status', methods=['GET'])
def get_system_status():
"""
Returns real-time system metadata including dynamic region and latency.
Accepts optional 'domain' parameter to check latency to a specific target.
"""
domain = request.args.get('domain')
stats = SystemMonitor.get_system_stats(target_url=domain)
return jsonify(stats)
@app.route('/api/seo-report', methods=['GET', 'POST'])
def get_seo_report():
# Cache variable attached to function to persist state
if not hasattr(get_seo_report, "cache"):
get_seo_report.cache = {"data": None, "mtime": 0}
def get_cached_report():
if not os.path.exists(REPORT_FILE):
return None
current_mtime = os.path.getmtime(REPORT_FILE)
if get_seo_report.cache["data"] is None or current_mtime > get_seo_report.cache["mtime"]:
try:
with open(REPORT_FILE, 'r', encoding='utf-8') as f:
get_seo_report.cache["data"] = json.load(f)
get_seo_report.cache["mtime"] = current_mtime
except Exception as e:
# If read fails, return None or raise
return None
return get_seo_report.cache["data"]
def get_param(name, default):
val = request.args.get(name) or request.form.get(name)
if val is None and request.is_json:
val = request.json.get(name)
return val if val is not None else default
# Global Caching for Domain Crawls (Active Memory Cache)
if not hasattr(get_seo_report, "domain_cache"):
get_seo_report.domain_cache = {}
domain = get_param('domain', None)
limit = int(get_param('limit', 25))
if domain:
# Check Cache (TTL 10 minutes)
cache_key = f"{domain}_{limit}"
cached_item = get_seo_report.domain_cache.get(cache_key)
import time
if cached_item:
timestamp, data = cached_item
# 600 seconds = 10 minutes
# DISABLE CACHE TEMPORARILY to ensure fresh code logic is used
# if time.time() - timestamp < 600:
# print(f"Returning Cached Result for {domain}")
# return jsonify(data)
try:
print(f"Starting live scan for: {domain}")
# Initialize Crawler
crawler = Crawler()
# Crawl the domain with the requested limit
site_data, total_discovered, _ = crawler.crawl_domain(domain, max_pages=limit)
if not site_data:
response = {
"summary": {
"total_pages_scanned": 0,
"total_images_found": 0,
"total_images_missing_alt": 0,
"total_pages_discovered": 0
},
"details": []
}
return jsonify(response)
# Analyze Results
analyzer = ImageAnalyzer()
results = analyzer.analyze_site(site_data)
# Add discovery stats
results['summary']['total_pages_discovered'] = total_discovered
results['details_count'] = len(results['details'])
# Save to Cache
get_seo_report.domain_cache[cache_key] = (time.time(), results)
return jsonify(results)
except Exception as e:
return jsonify({"error": f"Scraping failed: {str(e)}"}), 500
if not os.path.exists(REPORT_FILE):
return jsonify({"error": "Report file not found. Please run result logic first."}), 404
try:
data = get_cached_report()
if data is None:
# Fallback if file doesn't exist or read failed
# But if we are here, we passed the os.path.exists check earlier,
# so strictly speaking we should just handle the None case.
# The previous logic had a check for os.path.exists(REPORT_FILE) at line 59.
# We can retain that or rely on get_cached_report returning None.
return jsonify({"error": "Report file not found or unreadable."}), 404
# Get parameters for filtering existing report
# limit is already extracted above
random_param = str(get_param('random', 'true')).lower()
is_random = random_param == 'true'
summary = data.get('summary', {})
details = data.get('details', [])
# Filter details if limit is provided
if limit is not None and limit > 0:
if is_random and limit < len(details):
details = random.sample(details, limit)
else:
details = details[:limit]
response = {
"summary": summary,
"details_count": len(details), # Useful logic for client
"details": details
}
return jsonify(response)
except Exception as e:
return jsonify({"error": f"Failed to read report: {str(e)}"}), 500
if __name__ == '__main__':
# Run on 0.0.0.0 to be accessible if needed, default port 5000
app.run(debug=True, host='0.0.0.0', port=5050)
|