Hana Celeste commited on
Commit
7b6e408
·
verified ·
1 Parent(s): e0f9d33

Update app/gscode_logic.py

Browse files
Files changed (1) hide show
  1. app/gscode_logic.py +32 -24
app/gscode_logic.py CHANGED
@@ -1,6 +1,7 @@
1
  # gscode_logic.py
2
  import aiohttp
3
  from bs4 import BeautifulSoup
 
4
 
5
  GENSHIN_API = (
6
  "https://genshin-impact.fandom.com/api.php"
@@ -16,57 +17,64 @@ class GenshinCodeLogic:
16
 
17
  async def start(self):
18
  if not self.session:
19
- self.session = aiohttp.ClientSession()
 
 
 
 
20
 
21
  async def stop(self):
22
  if self.session:
23
  await self.session.close()
24
  self.session = None
25
 
26
- async def fetch_codes(self):
27
- if not self.session: await self.start()
28
-
29
  async with self.session.get(GENSHIN_API, timeout=aiohttp.ClientTimeout(total=20)) as resp:
30
  if resp.status != 200:
31
  return {"count": 0, "codes": []}
 
32
  data = await resp.json()
33
 
34
  html = data.get("parse", {}).get("text", {}).get("*")
35
- if not html: return {"count": 0, "codes": []}
 
36
 
37
  soup = BeautifulSoup(html, "html.parser")
38
- results = []
39
 
40
- rows = soup.select("table.article-table tr")
41
 
42
- for row in rows:
43
- cols = row.find_all("td")
44
- if len(cols) < 3: continue # Bỏ qua hàng tiêu đề hoặc hàng thiếu dữ liệu
 
 
 
 
 
45
 
46
- code_tag = cols[0].find("code")
47
- if not code_tag: continue
48
- code_text = code_tag.get_text(strip=True)
49
 
50
- server = cols[1].get_text(strip=True)
 
51
 
 
52
  rewards = []
53
- reward_items = cols[2].select(".item-text")
54
- for item in reward_items:
55
- rewards.append(item.get_text(strip=True))
56
-
57
- if not rewards:
58
- rewards = [cols[2].get_text(strip=True).replace("\n", ", ")]
59
 
60
- status_text = cols[3].get_text(" ", strip=True) if len(cols) > 3 else "Unknown"
 
61
 
62
  results.append({
63
- "code": code_text,
64
  "server": server,
65
  "rewards": rewards,
66
- "status": status_text
67
  })
68
 
69
  return {
70
  "count": len(results),
71
  "codes": results
72
- }
 
1
  # gscode_logic.py
2
  import aiohttp
3
  from bs4 import BeautifulSoup
4
+ from typing import List, Dict
5
 
6
  GENSHIN_API = (
7
  "https://genshin-impact.fandom.com/api.php"
 
17
 
18
  async def start(self):
19
  if not self.session:
20
+ self.session = aiohttp.ClientSession(
21
+ headers={
22
+ "User-Agent": "Mozilla/5.0 (GenshinCodeBot)"
23
+ }
24
+ )
25
 
26
  async def stop(self):
27
  if self.session:
28
  await self.session.close()
29
  self.session = None
30
 
31
+ async def fetch_codes(self) -> Dict:
 
 
32
  async with self.session.get(GENSHIN_API, timeout=aiohttp.ClientTimeout(total=20)) as resp:
33
  if resp.status != 200:
34
  return {"count": 0, "codes": []}
35
+
36
  data = await resp.json()
37
 
38
  html = data.get("parse", {}).get("text", {}).get("*")
39
+ if not html:
40
+ return {"count": 0, "codes": []}
41
 
42
  soup = BeautifulSoup(html, "html.parser")
 
43
 
44
+ results: List[Dict] = []
45
 
46
+ for row in soup.select("table.wikitable tbody tr"):
47
+ tds = row.find_all("td")
48
+ if len(tds) < 3:
49
+ continue
50
+
51
+ code_el = tds[0].find("code")
52
+ if not code_el:
53
+ continue
54
 
55
+ code = code_el.get_text(strip=True)
 
 
56
 
57
+ # server (China / Global / etc)
58
+ server = tds[1].get_text(" ", strip=True)
59
 
60
+ # rewards
61
  rewards = []
62
+ for item in tds[2].select(".item-text"):
63
+ text = item.get_text(" ", strip=True)
64
+ if text:
65
+ rewards.append(text)
 
 
66
 
67
+ # validity (nếu )
68
+ validity = tds[3].get_text(" ", strip=True) if len(tds) >= 4 else None
69
 
70
  results.append({
71
+ "code": code,
72
  "server": server,
73
  "rewards": rewards,
74
+ "validity": validity
75
  })
76
 
77
  return {
78
  "count": len(results),
79
  "codes": results
80
+ }