import http.server import socketserver import json import os import subprocess import threading from urllib.parse import urlparse, parse_qs PORT = int(os.environ.get('SERVER_PORT', 8000)) USE_HTTPS = os.environ.get('USE_HTTPS', 'false').lower() == 'true' SSL_CERT = '/etc/letsencrypt/live/stats.bacter1a.com/fullchain.pem' SSL_KEY = '/etc/letsencrypt/live/stats.bacter1a.com/privkey.pem' # server.pyの場所を基準に絶対パスを決定する BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DIRECTORY = BASE_DIR # 直接アクセスを禁止する機密ファイルリスト BLOCKED_PATHS = {'/settings.json', '/.gitignore', '/AI_RULES.md'} class DashboardHTTPRequestHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=DIRECTORY, **kwargs) def do_GET(self): """機密ファイルへの直接アクセスを禁止する""" parsed_path = urlparse(self.path) if parsed_path.path in BLOCKED_PATHS: self.send_response(403) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "error", "message": "Forbidden"}).encode('utf-8')) return super().do_GET() def do_POST(self): parsed_path = urlparse(self.path) # Endpoint for adding influencer if parsed_path.path == '/api/influencers': content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) try: new_influencer = json.loads(post_data.decode('utf-8')) # 1. config.json の更新 config_path = os.path.join(BASE_DIR, 'config.json') if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) else: config = {"influencers": []} # ID重複チェック new_id = new_influencer.get('id', '') existing_ids = [inf.get('id') for inf in config.get('influencers', [])] if new_id in existing_ids: self.send_response(409) self.send_header('Content-type', 'application/json') self.end_headers() error_response = {"status": "error", "message": f"ID '{new_id}' は既に登録されています"} self.wfile.write(json.dumps(error_response, ensure_ascii=False).encode('utf-8')) return config["influencers"].append(new_influencer) with open(config_path, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) # 2. main.py (データ収集) をバックグラウンドで実行 def run_scraper(): print(f"Triggering data collection for {new_influencer['name']}...") try: backend_dir = os.path.join(BASE_DIR, 'backend') subprocess.run(["python3", "main.py"], cwd=backend_dir, check=True) print("Data collection completed.") except subprocess.CalledProcessError as e: print(f"Error running scraper: {e}") threading.Thread(target=run_scraper).start() # 3. レスポンス返却 self.send_response(200) self.send_header('Content-type', 'application/json') self.end_headers() response = {"status": "success", "message": "Influencer added and collecting data in background"} self.wfile.write(json.dumps(response).encode('utf-8')) except Exception as e: print(f"Error handling POST /api/influencers: {e}") self.send_response(500) self.send_header('Content-type', 'application/json') self.end_headers() error_response = {"status": "error", "message": str(e)} self.wfile.write(json.dumps(error_response).encode('utf-8')) else: self.send_response(404) self.end_headers() def do_DELETE(self): parsed_path = urlparse(self.path) # DELETE /api/influencers?id=xxx if parsed_path.path == '/api/influencers': params = parse_qs(parsed_path.query) target_id = params.get('id', [None])[0] if not target_id: self.send_response(400) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "error", "message": "id parameter is required"}).encode('utf-8')) return try: config_path = os.path.join(BASE_DIR, 'config.json') if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) else: config = {"influencers": []} original_len = len(config.get("influencers", [])) config["influencers"] = [inf for inf in config.get("influencers", []) if inf.get("id") != target_id] if len(config["influencers"]) == original_len: self.send_response(404) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "error", "message": f"ID '{target_id}' not found"}).encode('utf-8')) return with open(config_path, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) self.send_response(200) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "success", "message": f"Influencer '{target_id}' deleted"}).encode('utf-8')) except Exception as e: print(f"Error handling DELETE /api/influencers: {e}") self.send_response(500) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"status": "error", "message": str(e)}).encode('utf-8')) else: self.send_response(404) self.end_headers() class ReusableTCPServer(socketserver.TCPServer): allow_reuse_address = True if __name__ == '__main__': with ReusableTCPServer(("", PORT), DashboardHTTPRequestHandler) as httpd: if USE_HTTPS: import ssl context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain(SSL_CERT, SSL_KEY) httpd.socket = context.wrap_socket(httpd.socket, server_side=True) print(f"Serving HTTPS at port {PORT}") print(f"Open https://stats.bacter1a.com/ in your browser.") else: print(f"Serving HTTP at port {PORT}") print(f"Open http://localhost:{PORT}/frontend/ in your browser.") httpd.serve_forever()