altrepo.parser: update news.news_urls parser

parent d37859f4
import aiohttp import aiohttp
from bs4 import BeautifulSoup
from datetime import datetime
import calendar
import re
from . import models from . import models
from .news import packages_parser from .news import urls_parser, packages_parser, bugs_parser
class BaseParser: class BaseParser:
...@@ -25,102 +20,14 @@ class NewsInfo: ...@@ -25,102 +20,14 @@ class NewsInfo:
self.client = client self.client = client
async def news_urls(self) -> models.NewsURL: async def news_urls(self) -> models.NewsURL:
now = datetime.now() return await urls_parser(self.client)
month_path = f"{now.year}-{calendar.month_name[now.month]}"
today_str = now.strftime("%Y%m%d")
base_url = f"https://lists.altlinux.org/pipermail/sisyphus-cybertalk/{month_path}/"
html = await self.client.get(f"{base_url}date.html", "koi8-r")
soup = BeautifulSoup(html, "html.parser")
sisyphus_id = 0
result = {"sisyphus": None, "bugs": None, "p11": None}
for li in soup.find_all("li"):
a = li.find("a")
if not a or not a.get("href"):
continue
href = a["href"]
text = a.get_text(strip=True)
match = re.search(r"(\d+)\.html", href)
if not match:
continue
msg_id = int(match.group(1))
url = base_url + href
if f"Sisyphus-{today_str} packages" in text:
result["sisyphus"] = url
sisyphus_id = msg_id
elif f"Sisyphus-{today_str} bugs" in text:
result["bugs"] = url
elif "p11/branch packages" in text and sisyphus_id:
if msg_id > sisyphus_id:
result["p11"] = url
if result["sisyphus"] is None:
result["p11"] = None
return models.NewsURL(**result)
async def bugs(self) -> models.BugsModel: async def bugs(self) -> models.BugsModel:
url = (await self.news_urls()).bugs url = (await self.news_urls()).bugs
if not url:
return None
html = await self.client.get(url, "koi8-r") html = await self.client.get(url, "koi8-r")
soup = BeautifulSoup(html, "html.parser") return await bugs_parser(html)
pre_tag = soup.find("pre")
if not pre_tag:
return {}
text = pre_tag.get_text()
lines = text.strip().splitlines()
data = {}
section_name = None
bug_pattern = re.compile(r"#(\d+)\s+([^\t]+)\s+([^\t]+)\s+([^\t]+)")
description_buffer = ""
current_bug = None
for line in lines:
header_match = re.match(r"^\s*(\d+)\s+(NEW|RESOLVED|REOPENED|RANDOM)\s+bugs?.*", line)
if header_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
current_bug = None
description_buffer = ""
section_name = await self._get_bug_section_name(line)
continue
bug_match = bug_pattern.match(line)
if bug_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
status = bug_match.group(4).strip()
if status == "---":
status = None
current_bug = {
"id": int(bug_match.group(1)),
"component": bug_match.group(2).strip(),
"priority": bug_match.group(3).strip(),
"status": status
}
description_buffer = ""
elif current_bug:
if not line.strip():
continue
if re.match(r"^Total\s+\d+\s+pending bugs", line.strip(), re.IGNORECASE):
continue
description_buffer += line.strip() + " "
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
return models.BugsModel(**data)
async def sisyphus(self) -> models.PackagesModel | None: async def sisyphus(self) -> models.PackagesModel | None:
url = (await self.news_urls()).sisyphus url = (await self.news_urls()).sisyphus
...@@ -135,21 +42,6 @@ class NewsInfo: ...@@ -135,21 +42,6 @@ class NewsInfo:
return None return None
html = await self.client.get(url, "koi8-r") html = await self.client.get(url, "koi8-r")
return await packages_parser(html) return await packages_parser(html)
async def _get_bug_section_name(self, line: str) -> str:
line = line.lower()
if "new" in line and "resolved" in line:
return "quickly_resolved"
elif "new" in line:
return "new"
elif "old" in line:
return "old"
elif "resolved" in line:
return "resolved"
elif "reopened" in line:
return "reopened"
elif "random" in line:
return "random"
class ALTRepoParser: class ALTRepoParser:
......
from .urls import urls_parser
from .packages import packages_parser from .packages import packages_parser
from .bugs import bugs_parser
from bs4 import BeautifulSoup
import re
from .. import models
async def bugs_parser(html: str):
soup = BeautifulSoup(html, "html.parser")
pre_tag = soup.find("pre")
if not pre_tag:
return {}
text = pre_tag.get_text()
lines = text.strip().splitlines()
data = {}
section_name = None
bug_pattern = re.compile(r"#(\d+)\s+([^\t]+)\s+([^\t]+)\s+([^\t]+)")
description_buffer = ""
current_bug = None
for line in lines:
header_match = re.match(
r"^\s*(\d+)\s+(NEW|RESOLVED|REOPENED|RANDOM)\s+bugs?.*", line)
if header_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
current_bug = None
description_buffer = ""
section_name = await _get_bug_section_name(line)
continue
bug_match = bug_pattern.match(line)
if bug_match:
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
status = bug_match.group(4).strip()
if status == "---":
status = None
current_bug = {
"id": int(bug_match.group(1)),
"component": bug_match.group(2).strip(),
"priority": bug_match.group(3).strip(),
"status": status
}
description_buffer = ""
elif current_bug:
if not line.strip():
continue
if re.match(r"^Total\s+\d+\s+pending bugs", line.strip(), re.IGNORECASE):
continue
description_buffer += line.strip() + " "
if current_bug and section_name:
current_bug["summary"] = description_buffer.strip()
data.setdefault(section_name, []).append(current_bug)
return models.BugsModel(**data)
async def _get_bug_section_name(line: str) -> str:
line = line.lower()
if "new" in line and "resolved" in line:
return "quickly_resolved"
elif "new" in line:
return "new"
elif "old" in line:
return "old"
elif "resolved" in line:
return "resolved"
elif "reopened" in line:
return "reopened"
elif "random" in line:
return "random"
from bs4 import BeautifulSoup
from datetime import datetime
import re
from .. import models
from config import CYBERTALK_URL
async def urls_parser(client):
now = datetime.now()
year_month = f"{now.year}-{now.strftime("%B")}"
today = now.strftime("%Y%m%d")
base_url = CYBERTALK_URL.format(year_month)
html = await client.get(f"{base_url}date.html", "koi8-r")
soup = BeautifulSoup(html, "html.parser")
sisyphus_id = 0
result = {"sisyphus": None, "bugs": None, "p11": None}
for li in soup.find_all("li"):
a = li.find("a")
if not a or not a.get("href"):
continue
href = a["href"]
text = a.get_text(strip=True)
match = re.search(r"(\d+)\.html", href)
if not match:
continue
msg_id = int(match.group(1))
url = base_url + href
if f"Sisyphus-{today} packages" in text:
result["sisyphus"] = url
sisyphus_id = msg_id
elif f"Sisyphus-{today} bugs" in text:
result["bugs"] = url
elif "p11/branch packages" in text and sisyphus_id:
if msg_id > sisyphus_id:
result["p11"] = url
if result["sisyphus"] is None:
result["p11"] = None
return models.NewsURL(**result)
...@@ -18,3 +18,5 @@ DEFAUIL_BRANCHES = [ ...@@ -18,3 +18,5 @@ DEFAUIL_BRANCHES = [
] ]
BUGS_URL = "https://bugzilla.altlinux.org/" BUGS_URL = "https://bugzilla.altlinux.org/"
PACKAGES_URL = "https://packages.altlinux.org/ru/{repo}/"
CYBERTALK_URL = "https://lists.altlinux.org/pipermail/sisyphus-cybertalk/{}/"
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment