Spaces:
Running
Running
File size: 4,827 Bytes
14b6a10 afe97a1 a38bf22 afe97a1 14b6a10 afe97a1 a38bf22 14b6a10 a38bf22 afe97a1 14b6a10 a38bf22 14b6a10 afe97a1 14b6a10 afe97a1 14b6a10 afe97a1 14b6a10 | 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 | """工具函數."""
from datetime import datetime
import pandas as pd
from .config import WORLDS
def format_price(price: int) -> str:
"""格式化價格顯示.
Args:
price: 價格數值
Returns:
格式化後的價格字串
"""
if price >= 1_000_000:
return f"{price / 1_000_000:.2f}M"
if price >= 1_000:
return f"{price / 1_000:.1f}K"
return str(price)
def format_timestamp(timestamp: int) -> str:
"""格式化時間戳.
Args:
timestamp: Unix 時間戳（秒或毫秒）
Returns:
格式化後的時間字串
"""
if not timestamp:
return "N/A"
# 如果 timestamp 超過合理範圍，可能是毫秒，需要轉換為秒
if timestamp > 9999999999:
timestamp = timestamp // 1000
dt = datetime.fromtimestamp(timestamp)
return dt.strftime("%Y-%m-%d %H:%M")
def format_relative_time(timestamp: int) -> str:
"""格式化相對時間.
Args:
timestamp: Unix 時間戳（秒或毫秒）
Returns:
相對時間字串（如「3 小時前」）
"""
if not timestamp:
return "N/A"
# 如果 timestamp 超過合理範圍，可能是毫秒，需要轉換為秒
if timestamp > 9999999999:
timestamp = timestamp // 1000
now = datetime.now()
dt = datetime.fromtimestamp(timestamp)
diff = now - dt
if diff.days > 0:
return f"{diff.days} 天前"
if diff.seconds >= 3600:
return f"{diff.seconds // 3600} 小時前"
if diff.seconds >= 60:
return f"{diff.seconds // 60} 分鐘前"
return "剛剛"
def process_listings(
listings: list,
quality: str = "all",
default_world: str = None,
retainer_filter: str = None,
) -> pd.DataFrame:
"""處理上架列表.
Args:
listings: 上架資料列表
quality: 品質篩選（"all", "nq", "hq"）
default_world: 預設伺服器名稱（當 API 未返回 worldID 時使用）
retainer_filter: 雇員名稱篩選（部分匹配，不區分大小寫）
Returns:
處理後的 DataFrame
"""
if not listings:
return pd.DataFrame()
data = []
for listing in listings:
if quality == "hq" and not listing.get("hq"):
continue
if quality == "nq" and listing.get("hq"):
continue
retainer_name = listing.get("retainerName", "")
# 雇員名稱篩選（部分匹配，不區分大小寫）
if retainer_filter:
if retainer_filter.lower() not in retainer_name.lower():
continue
# 優先使用 worldName，其次使用 worldID 對應，最後使用預設值
world_name = listing.get("worldName")
if not world_name:
world_id = listing.get("worldID")
if world_id:
world_name = WORLDS.get(world_id, str(world_id))
else:
world_name = default_world or "-"
data.append({
"品質": "HQ" if listing.get("hq") else "NQ",
"單價": listing.get("pricePerUnit", 0),
"數量": listing.get("quantity", 0),
"總價": listing.get("total", 0),
"雇員": retainer_name,
"伺服器": world_name,
"更新時間": format_relative_time(listing.get("lastReviewTime", 0)),
})
return pd.DataFrame(data)
def process_history(
entries: list,
quality: str = "all",
default_world: str = None,
) -> pd.DataFrame:
"""處理交易歷史.
Args:
entries: 交易記錄列表
quality: 品質篩選（"all", "nq", "hq"）
default_world: 預設伺服器名稱（當 API 未返回 worldID 時使用）
Returns:
處理後的 DataFrame
"""
if not entries:
return pd.DataFrame()
data = []
for entry in entries:
if quality == "hq" and not entry.get("hq"):
continue
if quality == "nq" and entry.get("hq"):
continue
# 優先使用 worldName，其次使用 worldID 對應，最後使用預設值
world_name = entry.get("worldName")
if not world_name:
world_id = entry.get("worldID")
if world_id:
world_name = WORLDS.get(world_id, str(world_id))
else:
world_name = default_world or "-"
data.append({
"品質": "HQ" if entry.get("hq") else "NQ",
"單價": entry.get("pricePerUnit", 0),
"數量": entry.get("quantity", 0),
"總價": entry.get("total", 0),
"買家": entry.get("buyerName", ""),
"伺服器": world_name,
"成交時間": format_timestamp(entry.get("timestamp", 0)),
})
return pd.DataFrame(data)
|