- 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>
125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
from datetime import timedelta
|
|
|
|
import redis
|
|
from dotenv import load_dotenv
|
|
from loguru import logger
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich import box
|
|
|
|
# 載入設定
|
|
load_dotenv()
|
|
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "192.168.42.211")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
|
|
REDIS_DB = int(os.getenv("REDIS_DB", 0))
|
|
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
|
|
|
|
TARGET_KEY = "pokemon:radar:latest"
|
|
|
|
def get_redis_client():
|
|
try:
|
|
r = redis.Redis(
|
|
host=REDIS_HOST,
|
|
port=REDIS_PORT,
|
|
db=REDIS_DB,
|
|
password=REDIS_PASSWORD,
|
|
decode_responses=True,
|
|
socket_connect_timeout=5
|
|
)
|
|
r.ping()
|
|
return r
|
|
except redis.RedisError as e:
|
|
logger.critical(f"Redis 連線失敗: {e}")
|
|
sys.exit(1)
|
|
|
|
def read_data():
|
|
r = get_redis_client()
|
|
|
|
if not r.exists(TARGET_KEY):
|
|
logger.warning(f"Key '{TARGET_KEY}' 不存在或已過期。")
|
|
return
|
|
|
|
ttl = r.ttl(TARGET_KEY)
|
|
raw_data = r.get(TARGET_KEY)
|
|
|
|
try:
|
|
data = json.loads(raw_data)
|
|
items = data.get("items", [])
|
|
|
|
console = Console()
|
|
|
|
# 顯示 Header 資訊
|
|
logger.info(f"成功讀取 Key: {TARGET_KEY}")
|
|
logger.info(f"剩餘存活時間 (TTL): {ttl} 秒 ({str(timedelta(seconds=ttl))})")
|
|
logger.success(f"總共發現 {len(items)} 隻寶可夢")
|
|
print() # 空一行
|
|
|
|
if not items:
|
|
return
|
|
|
|
# === 建立 Rich 表格 ===
|
|
table = Table(title="📡 寶可夢雷達掃描清單", box=box.ROUNDED)
|
|
|
|
# 定義欄位 (修正這裡)
|
|
# no_wrap 必須作為獨立參數傳入,不能寫在 style 字串裡
|
|
table.add_column("#", justify="right", style="cyan", no_wrap=True)
|
|
table.add_column("名稱 (性別)", style="bold white")
|
|
table.add_column("IV", justify="right", style="magenta")
|
|
table.add_column("CP / Lv", justify="right", style="yellow")
|
|
table.add_column("技能組合 (快/大)", style="dim")
|
|
table.add_column("剩餘時間", justify="right", style="green")
|
|
table.add_column("座標連結", style="blue underline")
|
|
|
|
for idx, item in enumerate(items):
|
|
# 1. 處理性別符號
|
|
gender_map = {"male": "♂", "female": "♀"}
|
|
gender = gender_map.get(item.get('gender'), "")
|
|
name_display = f"{item.get('name', 'Unknown')} {gender}"
|
|
|
|
# 2. 處理 IV (100% 亮紅顯示)
|
|
iv = item.get('iv_pct', 0)
|
|
iv_str = f"{iv}%"
|
|
if iv == 100:
|
|
iv_str = f"[bold red blink]{iv_str}[/]"
|
|
|
|
# 3. 處理數值
|
|
cp = item.get('cp', '?')
|
|
level = item.get('level', '?')
|
|
stats = f"CP {cp} / Lv {level}"
|
|
|
|
# 4. 處理技能
|
|
move1 = item.get('fast_move', '-')
|
|
move2 = item.get('charge_move', '-')
|
|
moves = f"{move1} / {move2}"
|
|
|
|
# 5. 處理時間
|
|
rem_sec = item.get('remaining', 0)
|
|
rem_str = f"{int(rem_sec // 60)}m {int(rem_sec % 60)}s"
|
|
|
|
# 6. 處理座標
|
|
lat = item.get('latitude')
|
|
lng = item.get('longitude')
|
|
coord_str = f"{lat:.5f}, {lng:.5f}"
|
|
|
|
table.add_row(
|
|
str(idx + 1),
|
|
name_display,
|
|
iv_str,
|
|
stats,
|
|
moves,
|
|
rem_str,
|
|
coord_str
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
except json.JSONDecodeError:
|
|
logger.error("JSON 解析失敗")
|
|
|
|
if __name__ == "__main__":
|
|
read_data()
|