- radar_fetcher.py:呼叫 Function Server 抓 twpkinfo.com 資料後寫入 Redis (TTL 300s) - pokemon_radar_api.py:FastAPI /scan,從 Redis 讀取並依 request 即時二次篩選;Redis 不可用時回 200 空結果,不再 500 - pokemon_location_fetcher.py:舊版 CLI,仍由 fetch_and_upload.sh 使用 - 文件:README (參考手冊)、QUICKSTART (操作指南)、SUMMARY (故事線)、CLAUDE.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
import json
|
||
import os
|
||
import redis
|
||
from dotenv import load_dotenv
|
||
|
||
# 載入 .env,確保跟原本的程式連到同一個 Redis
|
||
load_dotenv()
|
||
|
||
# Redis 設定 (從 env 讀取,預設值對齊 radar_fetcher.py)
|
||
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
|
||
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
|
||
REDIS_DB = int(os.getenv("REDIS_DB", 0))
|
||
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD") or None
|
||
REDIS_KEY = os.getenv("REDIS_KEY", "pokemon:radar:latest")
|
||
|
||
|
||
def main():
|
||
print(f"🔌 連線到 Redis: {REDIS_HOST}:{REDIS_PORT} (DB={REDIS_DB})")
|
||
|
||
try:
|
||
r = redis.Redis(
|
||
host=REDIS_HOST,
|
||
port=REDIS_PORT,
|
||
db=REDIS_DB,
|
||
password=REDIS_PASSWORD,
|
||
decode_responses=True,
|
||
socket_timeout=5,
|
||
)
|
||
|
||
# 測試連線
|
||
r.ping()
|
||
|
||
# 撈資料
|
||
print(f"🔍 正在查詢 Key: {REDIS_KEY} ...")
|
||
raw_data = r.get(REDIS_KEY)
|
||
|
||
if not raw_data:
|
||
print("❌ 找不到資料!Redis 裡沒有這個 Key,或是內容為空。")
|
||
return
|
||
|
||
# 解析並漂亮列印
|
||
try:
|
||
data = json.loads(raw_data)
|
||
|
||
# 顯示摘要資訊
|
||
print("-" * 40)
|
||
print(f"✅ 讀取成功!")
|
||
print(f"📅 資料產生時間: {data.get('generated_at', '未知')}")
|
||
print(f"🔢 原始數量: {data.get('count_raw', 'N/A')}")
|
||
print(f"🧹 過濾後數量: {data.get('count_filtered', 'N/A')}")
|
||
print("-" * 40)
|
||
print("📦 詳細內容 (JSON):")
|
||
print(json.dumps(data, indent=2, ensure_ascii=False))
|
||
print("-" * 40)
|
||
|
||
except json.JSONDecodeError:
|
||
print("⚠️ 資料格式錯誤!雖然有抓到字串,但不是合法的 JSON。")
|
||
print(f"內容: {raw_data}")
|
||
|
||
except Exception as e:
|
||
print(f"💥 發生錯誤: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|