vincent/publish-assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source 
Hugo/PaperMod static site tracking 13 conferences and 7 journals for
  edge and cloud systems research. Includes FullCalendar deadline view,
  ICORE/SCImago rankings, DBLP paper digest pipeline, and Python
  fetch/generate scripts. PaperMod added as a git submodule.
This commit is contained in:
khannurien
2026-04-24 11:52:33 +00:00
commit 8484abea47
57 changed files with 20183 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/publish_assistant/__init__.py Normal file
View File

202
src/publish_assistant/fetch_best_papers.py Normal file
View File

@@ -0,0 +1,202 @@
#!/usr/bin/env python3
"""
Fetch best-paper-award data from jeffhuang.com/best_paper_awards/.
The page lists best (and honourable mention) paper awards from ~32 CS venues
going back to 1996. It is updated manually by the maintainer.
HTML layout: one <table id="YEAR"> per year; inside each table each row has
<th class="category-name" rowspan=N> → venue name
<td> → paper title (as <a> link)
<td class="authors"> → author list
Output: site/data/best_papers.yaml (keyed by uppercase venue acronym)
OSDI:
- year: 2024
title: "Basilisk: Using Provenance Invariants ..."
authors: ["Tony Nuda Zhang", ...]
Downstream use: pa-generate reads this file to annotate digest pages with
a best-paper badge when a selected paper matches an award winner (matched
by case-insensitive title substring).
Usage:
uv run pa-fetch-best-papers [--output PATH]
"""
import argparse
import re
import sys
from pathlib import Path
import requests
import yaml
from bs4 import BeautifulSoup
SOURCE_URL = "https://jeffhuang.com/best_paper_awards/"
HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Accept-Language": "en-US,en;q=0.9",
}
# Maps the anchor name from jeffhuang.com → canonical acronym used in venues.yaml.
# The anchor appears as <a href="./conferences.html#<key>">
VENUE_MAP: dict[str, str] = {
"osdi": "OSDI",
"sosp": "SOSP",
"nsdi": "NSDI",
"eurosys": "EuroSys",
"usenix-atc": "ATC",
"atc": "ATC",
"socc": "SoCC",
"middleware": "Middleware",
"ipdps": "IPDPS",
"sc": "SC",
"hpdc": "HPDC",
"mobisys": "MobiSys",
"mobicom": "MobiCom",
"sigcomm": "SIGCOMM",
"sigmetrics": "SIGMETRICS",
"isca": "ISCA",
"micro": "MICRO",
"asplos": "ASPLOS",
"usenix-security": "USENIX Security",
"ccs": "CCS",
"sp": "S&P",
"ndss": "NDSS",
"fse": "FSE",
"icse": "ICSE",
"aaai": "AAAI",
"acl": "ACL",
"cvpr": "CVPR",
"focs": "FOCS",
"icml": "ICML",
"ijcai": "IJCAI",
"neurips": "NeurIPS",
"sigmod": "SIGMOD",
"vldb": "VLDB",
"www": "WWW",
"chi": "CHI",
"uist": "UIST",
"kdd": "KDD",
"sigir": "SIGIR",
"siggraph": "SIGGRAPH",
"stoc": "STOC",
"soda": "SODA",
}
def _extract_venue_key(th_tag) -> str | None:
"""Extract the venue from a <th class='category-name'> element."""
a = th_tag.find("a")
if not a:
return None
href = a.get("href", "")
m = re.search(r"#(.+)$", href)
if not m:
return None
anchor = m.group(1).lower().strip()
return VENUE_MAP.get(anchor)
def _extract_authors(td_tag) -> list[str]:
"""Extract author names from a <td class='authors'> element."""
# Collect all visible text nodes (hidden extra authors are in a div.d-none)
names: list[str] = []
# Primary text before the et-al link
raw = ""
for child in td_tag.children:
if hasattr(child, "get") and child.get("class") and "d-none" in child.get("class", []):
# Expanded authors div
for br_sibling in child.get_text("\n").split("\n"):
n = br_sibling.split(",")[0].strip()
if n and len(n) > 1:
names.append(n)
elif hasattr(child, "name") and child.name == "a":
continue # skip the "et al." toggle link
else:
raw += getattr(child, "string", child) or ""
# Parse primary visible text: "Name, Institution & Name2, Inst2" or "Name, Inst\nName2, Inst2"
if raw.strip():
for segment in re.split(r"[;&\n]|(?:\s*,\s*(?=[A-Z]))", raw):
n = segment.split(",")[0].strip()
if n and len(n) > 2 and not re.match(r"^(University|Institute|School|Dept|Lab|MIT|ETH|CMU|IBM|Google|Meta|Microsoft|Amazon|Apple|Intel|NVIDIA)", n):
names.append(n)
return names[:8] # cap at 8 to keep YAML manageable
def fetch_best_papers() -> dict[str, list[dict]]:
"""Return a dict keyed by venue acronym; each value is a list of award dicts."""
r = requests.get(SOURCE_URL, headers=HEADERS, timeout=30)
r.raise_for_status()
soup = BeautifulSoup(r.text, "lxml")
results: dict[str, list[dict]] = {}
# Each year is a <table id="YYYY">
for table in soup.find_all("table"):
table_id = table.get("id", "")
if not re.fullmatch(r"\d{4}", table_id):
continue
year = int(table_id)
current_venue: str | None = None
for row in table.find_all("tr"):
# Category header cell updates the current venue
cat_th = row.find("th", class_="category-name")
if cat_th:
current_venue = _extract_venue_key(cat_th)
if not current_venue:
continue
# Paper title cell
tds = row.find_all("td")
if not tds:
continue
title_td = tds[0]
link = title_td.find("a")
if not link:
continue
title = link.get_text(strip=True)
if len(title) < 8:
continue
# Authors cell
authors: list[str] = []
if len(tds) >= 2:
authors = _extract_authors(tds[1])
results.setdefault(current_venue, []).append({
"year": year,
"title": title,
"authors": authors,
"source": SOURCE_URL,
})
return results
def main() -> None:
ap = argparse.ArgumentParser(description="Fetch best-paper awards from jeffhuang.com.")
ap.add_argument("--output", default="site/data/best_papers.yaml")
args = ap.parse_args()
print(f"Fetching {SOURCE_URL} …", file=sys.stderr)
data = fetch_best_papers()
total = sum(len(v) for v in data.values())
print(f"Found {total} award entries across {len(data)} venues", file=sys.stderr)
for venue, entries in sorted(data.items()):
print(f" {venue}: {len(entries)}", file=sys.stderr)
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
with open(out, "w", encoding="utf-8") as f:
yaml.dump(data, f, allow_unicode=True, sort_keys=True, default_flow_style=False)
print(f"Written → {out}", file=sys.stderr)
if __name__ == "__main__":
main()

232
src/publish_assistant/fetch_deadlines.py Normal file
View File

@@ -0,0 +1,232 @@
#!/usr/bin/env python3
"""
Fetch submission deadlines for tracked conferences from WikiCFP.
For each conference in data/venues.yaml, this script:
1. Searches WikiCFP by acronym (or uses wikicfp_series if provided)
2. Scrapes the matching CFP page for deadline dates
3. Writes results to data/deadlines.yaml
Conferences where no CFP was found are listed under the 'missing' key in the
output YAML — the agent should fill those in manually (Task 2 in README).
Usage:
python fetch_deadlines.py [--venues PATH] [--output PATH]
"""
import argparse
import re
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
import requests
import yaml
from bs4 import BeautifulSoup
WIKICFP_SEARCH = "http://www.wikicfp.com/cfp/servlet/tool.search"
WIKICFP_EVENT = "http://www.wikicfp.com/cfp/servlet/event.showcfp"
WIKICFP_SERIES = "http://www.wikicfp.com/cfp/program?id={series_id}"
HEADERS = {
"User-Agent": "publish-assistant/1.0 (academic research tool)",
"Accept-Language": "en-US,en;q=0.9",
}
REQUEST_DELAY = 2.5
def search_wikicfp(session: requests.Session, query: str) -> list[dict]:
"""Search WikiCFP for a query string. Returns a list of CFP summary dicts."""
r = session.get(
WIKICFP_SEARCH,
params={"q": query, "year": "f"},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
soup = BeautifulSoup(r.text, "lxml")
results = []
# WikiCFP search results use paired rows: [name+link | when/where] [desc]
contsec = soup.find("div", class_="contsec")
if not contsec:
return results
rows = contsec.find_all("tr")
i = 0
while i < len(rows):
cells = rows[i].find_all("td")
# First row of a pair has 4 cells: title, acronym, deadline, event-date
if len(cells) >= 3:
link = cells[0].find("a", href=re.compile(r"showcfp\?eventid="))
if link:
m = re.search(r"eventid=(\d+)", link["href"])
event_id = m.group(1) if m else None
results.append({
"event_id": event_id,
"title": cells[0].get_text(strip=True),
"acronym": cells[1].get_text(strip=True) if len(cells) > 1 else "",
"deadline": cells[2].get_text(strip=True) if len(cells) > 2 else "",
"event_dates": cells[3].get_text(strip=True) if len(cells) > 3 else "",
})
i += 1
return results
def fetch_cfp_details(session: requests.Session, event_id: str) -> dict:
"""Scrape a WikiCFP event page for structured deadline information."""
r = session.get(
WIKICFP_EVENT,
params={"eventid": event_id},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
soup = BeautifulSoup(r.text, "lxml")
details: dict = {"wikicfp_event_id": event_id, "source": "wikicfp"}
# WikiCFP event pages use <th> for labels and <td> for values in the same <tr>
for row in soup.find_all("tr"):
th = row.find("th", recursive=False)
td = row.find("td", recursive=False)
if not (th and td):
continue
label = th.get_text(strip=True).lower()
value = td.get_text(" ", strip=True)
if not value or value in ("N/A", "TBD"):
continue
if "abstract" in label:
details["abstract_deadline"] = value
elif "submission" in label or "paper" in label:
details["submission_deadline"] = value
elif "notification" in label:
details["notification"] = value
elif "camera" in label or "final" in label:
details["camera_ready"] = value
elif "when" in label:
details["event_dates"] = value
elif "where" in label or "location" in label:
details["location"] = value
# Grab CFP link from the page header area
cfp_link = soup.find("a", href=re.compile(r"^https?://"), string=re.compile(r"cfp|call|submit", re.I))
if cfp_link and "cfp_url" not in details:
details["cfp_url"] = cfp_link["href"]
return details
def best_match(results: list[dict], acronym: str) -> dict | None:
"""Return the most likely match for a conference acronym from search results."""
acronym_up = acronym.upper()
# Require the acronym to appear as a standalone token (e.g. "SOSP 2026", not "EESP-SC")
pattern = re.compile(r"(?<![A-Z-])" + re.escape(acronym_up) + r"(?![A-Z])")
for r in results:
if pattern.search(r["title"].upper()):
return r
return None
def process_venue(session: requests.Session, conf: dict) -> dict | None:
"""Return deadline data for a conference, or None if not found."""
acronym = conf.get("acronym", "")
wikicfp_id = conf.get("wikicfp_id")
# wikicfp_id: false means explicitly not on WikiCFP (skip search)
if wikicfp_id is False:
return None
# wikicfp_id: "<number>" means use that event page directly
if wikicfp_id:
details = fetch_cfp_details(session, str(wikicfp_id))
details["wikicfp_title"] = acronym
return details
results = search_wikicfp(session, acronym)
match = best_match(results, acronym)
if not match or not match.get("event_id"):
return None
time.sleep(REQUEST_DELAY)
details = fetch_cfp_details(session, match["event_id"])
details["wikicfp_title"] = match.get("title", "")
return details
def main() -> None:
ap = argparse.ArgumentParser(description="Fetch submission deadlines from WikiCFP.")
ap.add_argument("--venues", default="site/data/venues.yaml")
ap.add_argument("--output", default="site/data/deadlines.yaml")
args = ap.parse_args()
venues_path = Path(args.venues)
if not venues_path.exists():
print(f"Error: {venues_path} not found. Populate data/venues.yaml first.", file=sys.stderr)
sys.exit(1)
with open(venues_path) as f:
venues = yaml.safe_load(f) or {}
conferences = venues.get("conferences") or []
if not conferences:
print("No conferences in venues.yaml. Nothing to do.", file=sys.stderr)
sys.exit(0)
# Preserve any manually-entered deadlines from the previous output
existing: dict[str, dict] = {}
out_path = Path(args.output)
if out_path.exists():
with open(out_path) as f:
prev = yaml.safe_load(f) or {}
for acronym, data in (prev.get("deadlines") or {}).items():
if (data or {}).get("source") == "manual":
existing[acronym] = data
session = requests.Session()
found: dict[str, dict] = dict(existing) # start with manual entries
missing: list[str] = []
for conf in conferences:
acronym = conf.get("acronym", "")
print(f" {acronym} …", file=sys.stderr)
if acronym in existing:
print(f" manual entry preserved", file=sys.stderr)
continue
try:
time.sleep(REQUEST_DELAY)
data = process_venue(session, conf)
if data:
found[acronym] = data
dl = data.get("submission_deadline") or data.get("abstract_deadline") or "?"
print(f" deadline: {dl}", file=sys.stderr)
else:
missing.append(acronym)
print(f" not found on WikiCFP", file=sys.stderr)
except Exception as exc:
missing.append(acronym)
print(f" error: {exc}", file=sys.stderr)
output = {
"generated": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"deadlines": found,
"missing": missing,
}
out_path = Path(args.output)
out_path.parent.mkdir(parents=True, exist_ok=True)
with open(out_path, "w") as f:
yaml.dump(output, f, allow_unicode=True, sort_keys=False, default_flow_style=False)
print(
f"\nResults: {len(found)} found, {len(missing)} missing → {out_path}",
file=sys.stderr,
)
if missing:
print(
f"Missing (fill in manually — see README Task 2): {', '.join(missing)}",
file=sys.stderr,
)
if __name__ == "__main__":
main()

143
src/publish_assistant/fetch_icore.py Normal file
View File

@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""
Fetch ICORE conference rankings from portal.core.edu.au and save to CSV.
Scrapes the portal's HTML search results with pagination. Results are merged
with any existing CSV so manual overrides survive.
Usage:
python fetch_icore.py [--query QUERY] [--source SOURCE] [--output PATH]
Examples:
# Fetch all rankings for the current default source
python fetch_icore.py --output data/rankings/icore.csv
# Fetch rankings matching a keyword
python fetch_icore.py --query "distributed" --output data/rankings/icore.csv
# Use an older ranking round
python fetch_icore.py --source CORE2021 --output data/rankings/icore_2021.csv
Available sources (as of 2024): CORE2023, CORE2021, CORE2020, CORE2018, CORE2017
"""
import argparse
import csv
import sys
import time
from pathlib import Path
import requests
from bs4 import BeautifulSoup
BASE_URL = "https://portal.core.edu.au/conf-ranks/"
HEADERS = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0"}
REQUEST_DELAY = 1.5 # seconds between paginated requests
def fetch_page(session: requests.Session, query: str, source: str, page: int) -> BeautifulSoup:
r = session.get(
BASE_URL,
params={"search": query, "by": "all", "source": source, "sort": "arank", "page": page},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
return BeautifulSoup(r.text, "lxml")
def parse_rows(soup: BeautifulSoup) -> list[dict]:
table = soup.find("table")
if not table:
return []
headers = [th.get_text(strip=True) for th in table.find_all("th")]
if not headers:
return []
rows = []
for tr in table.find_all("tr"):
cells = [td.get_text(strip=True) for td in tr.find_all("td")]
if len(cells) == len(headers):
rows.append(dict(zip(headers, cells)))
return rows
def get_total_pages(soup: BeautifulSoup) -> int:
import re
# Portal uses javascript: jumpPage('N') links for pagination
max_page = 1
for a in soup.find_all("a", href=re.compile(r"jumpPage")):
m = re.search(r"jumpPage\('?(\d+)'?\)", a.get("href", ""))
if m:
max_page = max(max_page, int(m.group(1)))
if max_page > 1:
return max_page
# Fallback: "Page X of Y" text
for s in soup.stripped_strings:
parts = s.strip().split()
if len(parts) >= 4 and parts[0] == "Page" and parts[2] == "of":
try:
return int(parts[3])
except ValueError:
pass
return 1
def fetch_all(query: str, source: str) -> list[dict]:
session = requests.Session()
print(f"Fetching ICORE rankings (source={source}, query={query!r}) …", file=sys.stderr)
first_page = fetch_page(session, query, source, 1)
rows = parse_rows(first_page)
total = get_total_pages(first_page)
print(f" {total} page(s) found", file=sys.stderr)
for page in range(2, total + 1):
print(f" page {page}/{total} …", file=sys.stderr)
time.sleep(REQUEST_DELAY)
rows.extend(parse_rows(fetch_page(session, query, source, page)))
return rows
def merge_and_write(new_rows: list[dict], output_path: Path) -> None:
"""Merge new_rows with existing CSV, keyed on Acronym. New data wins."""
existing: dict[str, dict] = {}
if output_path.exists():
with open(output_path, newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
key = row.get("Acronym") or row.get("acronym") or ""
existing[key] = row
for row in new_rows:
key = row.get("Acronym") or row.get("acronym") or ""
existing[key] = row
if not existing:
print("No data to write.", file=sys.stderr)
return
output_path.parent.mkdir(parents=True, exist_ok=True)
fieldnames = list(next(iter(existing.values())).keys())
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(existing.values())
print(f"Saved {len(existing)} entries → {output_path}", file=sys.stderr)
def main() -> None:
ap = argparse.ArgumentParser(description="Fetch ICORE conference rankings.")
ap.add_argument("--query", default="", help="Search term (default: all)")
ap.add_argument("--source", default="CORE2023", help="Ranking source (default: CORE2023)")
ap.add_argument("--output", default="site/data/rankings/icore.csv", help="Output CSV path")
args = ap.parse_args()
rows = fetch_all(args.query, args.source)
if not rows:
print(
"No rows returned. The portal may have changed its HTML structure, "
"or the search returned no results.",
file=sys.stderr,
)
sys.exit(1)
merge_and_write(rows, Path(args.output))
if __name__ == "__main__":
main()

256
src/publish_assistant/fetch_papers.py Normal file
View File

@@ -0,0 +1,256 @@
#!/usr/bin/env python3
"""
Fetch paper metadata for a conference or journal by venue acronym and year.
Primary source: DBLP (most reliable for CS venues).
Fallback: OpenAlex (broader coverage, includes citation counts).
Output is a YAML candidate pool at data/papers/<VENUE>-<YEAR>-candidates.yaml.
The agent then selects notable papers from this pool and writes a digest YAML
(data/papers/<VENUE>-<YEAR>-digest.yaml) — see README Task 3.
Usage:
python fetch_papers.py --venue OSDI --year 2025
python fetch_papers.py --venue OSDI --year 2025 --dblp-key conf/osdi
python fetch_papers.py --venue TPDS --year 2024 --source openalex
Tips:
- If DBLP auto-detection fails, look up the key at https://dblp.org/db/conf/
or https://dblp.org/db/journals/ and pass it with --dblp-key.
- For journals, the DBLP key is usually like "journals/tpds".
- For conferences, it is usually like "conf/osdi" or "conf/eurosys".
"""
import argparse
import sys
import time
from pathlib import Path
import requests
import yaml
DBLP_SEARCH_URL = "https://dblp.org/search/publ/api"
DBLP_VENUE_URL = "https://dblp.org/search/venue/api"
OPENALEX_WORKS_URL = "https://api.openalex.org/works"
OPENALEX_VENUES_URL = "https://api.openalex.org/venues"
HEADERS = {
"User-Agent": "publish-assistant/1.0 (academic research tool; mailto:research@example.org)"
}
REQUEST_DELAY = 1.0
# --- DBLP helpers ---
def dblp_find_venue_key(acronym: str) -> str | None:
"""Search DBLP for a venue and return its stream key (e.g. 'conf/osdi')."""
r = requests.get(
DBLP_VENUE_URL,
params={"q": acronym, "format": "json", "h": 20},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
hits = r.json().get("result", {}).get("hits", {}).get("hit", []) or []
acronym_up = acronym.upper()
for hit in hits:
info = hit.get("info", {})
url = info.get("url", "")
name = info.get("venue", "")
# Match acronym in the venue URL path segment or venue name
if acronym_up in name.upper() or f"/{acronym.lower()}/" in url.lower() or url.lower().endswith(f"/{acronym.lower()}"):
if "/db/" in url:
return url.split("/db/")[-1].rstrip("/")
return None
def dblp_fetch_papers(venue_key: str, year: int, max_results: int = 500) -> list[dict]:
"""Fetch papers from DBLP for a venue key and year using the search API."""
papers: list[dict] = []
batch = 250
offset = 0
query = f"streamid:{venue_key}: year:{year}"
while len(papers) < max_results:
r = requests.get(
DBLP_SEARCH_URL,
params={"q": query, "format": "json", "h": batch, "f": offset},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
result = r.json().get("result", {})
hits = result.get("hits", {}).get("hit", []) or []
if not hits:
break
for hit in hits:
info = hit.get("info", {})
# Authors may be a string, dict, or list depending on count
raw_authors = info.get("authors", {}).get("author", []) or []
if isinstance(raw_authors, str):
authors = [raw_authors]
elif isinstance(raw_authors, dict):
authors = [raw_authors.get("text", "")]
else:
authors = [
a.get("text", a) if isinstance(a, dict) else str(a)
for a in raw_authors
]
papers.append({
"title": (info.get("title") or "").rstrip("."),
"authors": authors,
"year": info.get("year"),
"doi": info.get("doi"),
"url": info.get("ee") or info.get("url"),
"dblp_key": info.get("key"),
"venue_name": info.get("venue"),
"pages": info.get("pages"),
})
offset += len(hits)
if len(hits) < batch:
break
time.sleep(REQUEST_DELAY)
return papers
# --- OpenAlex helpers ---
def openalex_find_venue_id(name: str) -> str | None:
"""Find an OpenAlex venue/source ID by display name."""
r = requests.get(
OPENALEX_VENUES_URL,
params={"search": name, "per_page": 5},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
results = r.json().get("results", [])
return results[0]["id"] if results else None
def openalex_fetch_papers(venue_name: str, year: int, max_results: int = 500) -> list[dict]:
"""Fetch papers from OpenAlex for a venue name and year."""
venue_id = openalex_find_venue_id(venue_name)
if not venue_id:
print(f" OpenAlex: no venue found for {venue_name!r}", file=sys.stderr)
return []
papers: list[dict] = []
page = 1
while len(papers) < max_results:
r = requests.get(
OPENALEX_WORKS_URL,
params={
"filter": f"primary_location.source.id:{venue_id},publication_year:{year}",
"per_page": 200,
"page": page,
"mailto": "research@example.org",
},
headers=HEADERS,
timeout=30,
)
r.raise_for_status()
results = r.json().get("results", [])
if not results:
break
for work in results:
authors = [
a.get("author", {}).get("display_name", "")
for a in work.get("authorships", [])
]
oa = work.get("open_access", {})
papers.append({
"title": work.get("title", ""),
"authors": authors,
"year": work.get("publication_year"),
"doi": work.get("doi"),
"url": work.get("primary_location", {}).get("landing_page_url"),
"open_access_url": oa.get("oa_url"),
"cited_by_count": work.get("cited_by_count", 0),
"openalex_id": work.get("id"),
})
if len(results) < 200:
break
page += 1
time.sleep(REQUEST_DELAY)
return papers
# --- Main ---
def main() -> None:
ap = argparse.ArgumentParser(description="Fetch paper candidates for a venue and year.")
ap.add_argument("--venue", required=True, help="Venue acronym (e.g. OSDI, EuroSys, TPDS)")
ap.add_argument("--year", required=True, type=int)
ap.add_argument(
"--output",
help="Output path (default: data/papers/<VENUE>-<YEAR>-candidates.yaml)",
)
ap.add_argument(
"--source",
choices=["dblp", "openalex", "auto"],
default="auto",
help="Data source (default: auto — tries DBLP first, then OpenAlex)",
)
ap.add_argument(
"--dblp-key",
help="Override DBLP venue key (e.g. conf/osdi). Use when auto-detection fails.",
)
ap.add_argument(
"--max", type=int, default=500, dest="max_results", help="Max papers to fetch (default: 500)"
)
args = ap.parse_args()
out_path = Path(args.output or f"site/data/papers/{args.venue}-{args.year}-candidates.yaml")
papers: list[dict] = []
source_used = args.source
if args.source in ("dblp", "auto"):
venue_key = args.dblp_key
if not venue_key:
print(f"Looking up DBLP key for {args.venue} …", file=sys.stderr)
venue_key = dblp_find_venue_key(args.venue)
if venue_key:
print(f"Fetching from DBLP (key: {venue_key}) …", file=sys.stderr)
papers = dblp_fetch_papers(venue_key, args.year, args.max_results)
source_used = "dblp"
else:
print(
f"Could not find DBLP key for {args.venue!r}. "
"Try --dblp-key or --source openalex.",
file=sys.stderr,
)
if not papers and args.source in ("openalex", "auto"):
print(f"Fetching from OpenAlex …", file=sys.stderr)
papers = openalex_fetch_papers(args.venue, args.year, args.max_results)
source_used = "openalex"
if not papers:
print(
"No papers found. "
"Check the acronym, try --dblp-key, or use --source openalex.",
file=sys.stderr,
)
sys.exit(1)
output = {
"venue": args.venue,
"year": args.year,
"source": source_used,
"count": len(papers),
"papers": papers,
}
out_path.parent.mkdir(parents=True, exist_ok=True)
with open(out_path, "w", encoding="utf-8") as f:
yaml.dump(output, f, allow_unicode=True, sort_keys=False, default_flow_style=False)
print(f"Saved {len(papers)} papers → {out_path}", file=sys.stderr)
print("Next step: review the candidates and create a digest YAML (see README Task 3).", file=sys.stderr)
if __name__ == "__main__":
main()

117
src/publish_assistant/fetch_scimago.py Normal file
View File

@@ -0,0 +1,117 @@
#!/usr/bin/env python3
"""
Download SCImago journal rankings as CSV.
SCImago publishes rankings by subject area. This script downloads the CSV for
a given area code and year. Use --list-areas to see available CS area codes.
Usage:
python fetch_scimago.py [--area CODE] [--year YEAR] [--output PATH]
python fetch_scimago.py --list-areas
Examples:
# Fetch all CS journals (area 1700), latest available year
python fetch_scimago.py --output data/rankings/scimago.csv
# Fetch only "Computer Networks and Communications" journals
python fetch_scimago.py --area 1705 --output data/rankings/scimago_networks.csv
# Fetch a specific year
python fetch_scimago.py --area 1700 --year 2022 --output data/rankings/scimago_2022.csv
"""
import argparse
import sys
from pathlib import Path
import requests
SCIMAGO_URL = "https://www.scimagojr.com/journalrank.php"
HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.5",
"Referer": "https://www.scimagojr.com/journalrank.php",
}
# Subject area codes for Computer Science and related fields
CS_AREAS: dict[str, str] = {
"1700": "Computer Science (general)",
"1701": "Computer Science Applications",
"1702": "Artificial Intelligence",
"1703": "Computational Theory and Mathematics",
"1704": "Computer Graphics and Computer-Aided Design",
"1705": "Computer Networks and Communications",
"1706": "Computer Science Miscellaneous",
"1707": "Computer Vision and Pattern Recognition",
"1708": "Hardware and Architecture",
"1709": "Human-Computer Interaction",
"1710": "Information Systems",
"1711": "Signal Processing",
"2200": "Engineering (general)",
"2202": "Aerospace Engineering",
"2208": "Electrical and Electronic Engineering",
}
def list_areas() -> None:
print("SCImago subject area codes (Computer Science / Engineering):\n")
for code, name in CS_AREAS.items():
print(f" {code} {name}")
def download_csv(area: str, year: str) -> str:
params = {
"area": area,
"category": 0,
"country": "all",
"year": year,
"order": "sjr",
"min": 0,
"min_type": "cd",
"type": "j",
"format": "csv",
}
label = CS_AREAS.get(str(area), area)
print(f"Downloading SCImago CSV: area={area} ({label}), year={year} …", file=sys.stderr)
session = requests.Session()
# First visit the page to acquire a session cookie
session.get(SCIMAGO_URL, headers=HEADERS, timeout=30)
r = session.get(SCIMAGO_URL, params=params, headers=HEADERS, timeout=60)
r.raise_for_status()
# SCImago may return an HTML anti-bot challenge instead of CSV
text = r.text.lstrip("")
if text.lstrip().startswith("<!"):
raise ValueError(
"SCImago returned an HTML page instead of CSV — likely blocked by anti-bot "
"protection. Download the CSV manually from scimagojr.com (Journal Rankings → "
"choose area → Download) and save it to site/data/rankings/scimago.csv.\n"
"Alternatively, add scimago_quartile / scimago_sjr fields directly to "
"site/data/venues.yaml (see existing journal entries for the format)."
)
if not text.strip():
raise ValueError("Empty response from SCImago. The area code or year may be invalid.")
return text
def main() -> None:
ap = argparse.ArgumentParser(description="Download SCImago journal rankings.")
ap.add_argument("--area", default="1700", help="Subject area code (default: 1700 — all CS)")
ap.add_argument("--year", default="2023", help="Ranking year (default: 2023)")
ap.add_argument("--output", default="site/data/rankings/scimago.csv")
ap.add_argument("--list-areas", action="store_true", help="Print available area codes and exit")
args = ap.parse_args()
if args.list_areas:
list_areas()
return
csv_text = download_csv(args.area, args.year)
out = Path(args.output)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(csv_text, encoding="utf-8")
n = csv_text.count("\n") - 1
print(f"Saved {n} entries → {out}", file=sys.stderr)
if __name__ == "__main__":
main()

606
src/publish_assistant/generate_content.py Normal file
View File

@@ -0,0 +1,606 @@
#!/usr/bin/env python3
"""
Generate Hugo content pages from data files.
Each page has a minimal front matter block (for Hugo taxonomy / metadata) plus
a rendered Markdown body so generic themes display content without custom templates.
Reads:
site/data/venues.yaml master venue list
site/data/rankings/icore.csv ICORE conference ranks
site/data/rankings/scimago.csv SCImago journal ranks (optional)
site/data/deadlines.yaml deadline data
site/data/papers/<V>-<Y>-candidates.yaml paper pools
site/data/papers/<V>-<Y>-digest.yaml agent-curated selections
Writes:
site/content/venues/conferences/<acronym>/_index.md
site/content/venues/journals/<acronym>/_index.md
site/content/calendar/_index.md
site/content/digests/<VENUE>-<YEAR>/index.md
PRESERVED_VENUE_FIELDS are never overwritten once set (manual edits survive).
"""
import argparse
import csv
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
import yaml
PRESERVED_VENUE_FIELDS = {"notes", "deadline_source"}
# Fields removed from existing front matter on every regeneration:
# - "url" reserved by Hugo (overrides page URL)
# - "papers" moved from front matter to rendered markdown body
_STRIP_FROM_EXISTING = {"url", "papers"}
# ---------------------------------------------------------------------------
# Front matter I/O
# ---------------------------------------------------------------------------
def read_hugo_file(path: Path) -> tuple[dict, str]:
if not path.exists():
return {}, ""
text = path.read_text(encoding="utf-8")
if text.startswith("---"):
parts = text.split("---", 2)
if len(parts) == 3:
return yaml.safe_load(parts[1]) or {}, parts[2]
return {}, text
def write_hugo_file(path: Path, fm: dict, body: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fm_str = yaml.dump(fm, allow_unicode=True, sort_keys=False, default_flow_style=False)
path.write_text(f"---\n{fm_str}---\n{body}", encoding="utf-8")
def merge_fm(existing: dict, new_data: dict, preserved: set[str]) -> dict:
result = {k: v for k, v in existing.items() if k not in _STRIP_FROM_EXISTING}
for key, value in new_data.items():
if key in preserved and result.get(key):
continue
result[key] = value
return result
# ---------------------------------------------------------------------------
# Data loaders
# ---------------------------------------------------------------------------
def load_icore(path: str) -> dict[str, str]:
p = Path(path)
if not p.exists():
return {}
ranks: dict[str, str] = {}
with open(p, newline="", encoding="utf-8") as f:
sample = f.read(4096); f.seek(0)
sep = ";" if sample.count(";") > sample.count(",") else ","
for row in csv.DictReader(f, delimiter=sep):
acronym = (row.get("Acronym") or row.get("acronym") or "").strip()
rank = (row.get("Rank") or row.get("rank") or row.get("CORE Rank") or "").strip()
if acronym:
ranks[acronym.upper()] = rank
return ranks
def load_scimago(path: str) -> dict[str, dict]:
p = Path(path)
if not p.exists():
return {}
journals: dict[str, dict] = {}
with open(p, newline="", encoding="utf-8") as f:
sample = f.read(4096); f.seek(0)
sep = ";" if sample.count(";") > sample.count(",") else ","
for row in csv.DictReader(f, delimiter=sep):
title = (row.get("Title") or row.get("title") or "").strip().lower()
if title:
journals[title] = {
"scimago_quartile": (row.get("SJR Best Quartile") or row.get("Best Quartile") or "").strip(),
"scimago_sjr": (row.get("SJR") or "").strip(),
"scimago_h_index": (row.get("H index") or row.get("H Index") or "").strip(),
}
return journals
def scimago_lookup(data: dict[str, dict], full_name: str, issn: str) -> dict:
entry = data.get(full_name.lower())
if entry:
return entry
needle = full_name.lower()
for title, entry in data.items():
if needle in title or title in needle:
return entry
return {}
def load_deadlines(path: str) -> dict[str, dict]:
p = Path(path)
if not p.exists():
return {}
with open(p) as f:
data = yaml.safe_load(f) or {}
return data.get("deadlines", {})
# ---------------------------------------------------------------------------
# Markdown body renderers
# ---------------------------------------------------------------------------
def _conf_body(fm: dict) -> str:
rank = fm.get("icore_rank") or "—"
hp = fm.get("homepage") or ""
dl = fm.get("next_deadline") or "TBD"
abstract_dl = fm.get("abstract_deadline") or ""
event = fm.get("next_event") or "—"
domains = ", ".join(f"`{d}`" for d in (fm.get("domain") or []))
notes = fm.get("notes") or ""
src = fm.get("deadline_source") or ""
lines = [f"\n*{fm.get('full_name', '')}*\n"]
lines += [
"| | |", "|---|---|",
f"| **ICORE Rank** | **{rank}** |",
f"| **Domains** | {domains} |",
]
if hp:
lines.append(f"| **Website** | [{hp}]({hp}) |")
lines += [
"\n## Upcoming Deadline\n",
"| | |", "|---|---|",
]
if abstract_dl and abstract_dl != dl:
lines.append(f"| **Abstract deadline** | {abstract_dl} |")
lines += [
f"| **Paper deadline** | {dl} |",
f"| **Event dates** | {event} |",
]
if src:
lines.append(f"| **Source** | `{src}` |")
if notes:
lines += ["\n## Notes\n", notes]
return "\n".join(lines) + "\n"
def _journal_body(fm: dict) -> str:
q = fm.get("scimago_quartile") or "—"
sjr = fm.get("scimago_sjr") or "—"
h = fm.get("scimago_h_index") or "—"
issn = fm.get("issn") or "—"
hp = fm.get("homepage") or ""
model = (fm.get("submission_model") or "rolling").title()
domains = ", ".join(f"`{d}`" for d in (fm.get("domain") or []))
notes = fm.get("notes") or ""
lines = [f"\n*{fm.get('full_name', '')}*\n"]
lines += [
"| | |", "|---|---|",
f"| **SCImago Quartile** | **{q}** |",
f"| **SJR** | {sjr} |",
f"| **H-Index** | {h} |",
f"| **ISSN** | {issn} |",
f"| **Submission model** | {model} |",
f"| **Domains** | {domains} |",
]
if hp:
lines.append(f"| **Website** | [{hp}]({hp}) |")
if notes:
lines += ["\n## Notes\n", notes]
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Date helpers
# ---------------------------------------------------------------------------
def _parse_date(s: str) -> str | None:
"""'Apr 1, 2026' → '2026-04-01', or None if unparseable."""
for fmt in ("%b %d, %Y", "%B %d, %Y"):
try:
return datetime.strptime(s.strip(), fmt).strftime("%Y-%m-%d")
except ValueError:
pass
return None
def _parse_date_range(s: str) -> tuple[str | None, str | None]:
"""Return (start_iso, end_iso_exclusive) from a date-range string.
Handles:
'Apr 13, 2026 - Apr 16, 2026' → ('2026-04-13', '2026-04-17')
'Nov 17-19, 2026' → ('2026-11-17', '2026-11-20')
'Apr 1, 2026' → ('2026-04-01', None)
"""
s = s.strip()
if " - " in s:
left, right = s.split(" - ", 1)
start = _parse_date(left)
end_s = _parse_date(right)
if start and end_s:
end_dt = datetime.strptime(end_s, "%Y-%m-%d") + timedelta(days=1)
return start, end_dt.strftime("%Y-%m-%d")
m = re.match(r"(\w+)\s+(\d+)-(\d+),\s+(\d{4})", s)
if m:
month, d1, d2, year = m.groups()
start = _parse_date(f"{month} {d1}, {year}")
try:
end_dt = datetime.strptime(f"{month} {d2}, {year}", "%b %d, %Y") + timedelta(days=1)
if start:
return start, end_dt.strftime("%Y-%m-%d")
except ValueError:
pass
return _parse_date(s), None
# ---------------------------------------------------------------------------
# Venue section index body renderers
# ---------------------------------------------------------------------------
_ICORE_ORDER = {"A*": 0, "A": 1, "B": 2, "C": 3}
_QUARTILE_ORDER = {"Q1": 0, "Q2": 1, "Q3": 2, "Q4": 3}
def _venues_body(venues: dict) -> str:
n_conf = len(venues.get("conferences") or [])
n_jour = len(venues.get("journals") or [])
return (
f"\nTracking **{n_conf} conferences** and **{n_jour} journals** for this domain.\n\n"
"- **[Conferences →](/venues/conferences/)** ranked by ICORE (A*, A, B, C)\n"
"- **[Journals →](/venues/journals/)** ranked by SCImago quartile (Q1–Q4)\n"
)
def _conferences_section_body(venues: dict, icore: dict, deadlines: dict) -> str:
rows = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
rank = icore.get(acronym.upper(), "—") or "—"
dl = deadlines.get(acronym, {})
deadline = dl.get("submission_deadline") or dl.get("abstract_deadline") or "TBD"
domains = ", ".join(f"`{d}`" for d in conf.get("domain", []))
rows.append((_ICORE_ORDER.get(rank, 99), acronym, conf.get("full_name", ""), rank, deadline, domains))
rows.sort(key=lambda r: (r[0], r[1]))
lines = ["\n| Conference | Full name | ICORE | Next deadline | Domains |",
"|---|---|---|---|---|"]
for _, acronym, full_name, rank, deadline, domains in rows:
link = f"[{acronym}](/venues/conferences/{acronym.lower()}/)"
lines.append(f"| {link} | {full_name} | **{rank}** | {deadline} | {domains} |")
return "\n".join(lines) + "\n"
def _journals_section_body(venues: dict) -> str:
rows = []
for j in venues.get("journals") or []:
acronym = j.get("acronym", "")
q = j.get("scimago_quartile") or "—"
sjr = j.get("scimago_sjr") or "—"
model = (j.get("submission_model") or "rolling").title()
domains = ", ".join(f"`{d}`" for d in j.get("domain", []))
rows.append((_QUARTILE_ORDER.get(q, 99), acronym, j.get("full_name", ""), q, sjr, model, domains))
rows.sort(key=lambda r: (r[0], r[1]))
lines = ["\n| Journal | Full name | SCImago | SJR | Model | Domains |",
"|---|---|---|---|---|---|"]
for _, acronym, full_name, q, sjr, model, domains in rows:
link = f"[{acronym}](/venues/journals/{acronym.lower()}/)"
lines.append(f"| {link} | {full_name} | **{q}** | {sjr} | {model} | {domains} |")
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Calendar events builder (for FullCalendar)
# ---------------------------------------------------------------------------
def _calendar_events(venues: dict, deadlines: dict) -> list[dict]:
events: list[dict] = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
dl = deadlines.get(acronym, {})
url = f"/venues/conferences/{acronym.lower()}/"
for field, label, color in [
("abstract_deadline", "abstract deadline", "#f4a261"),
("submission_deadline", "paper deadline", "#e63946"),
]:
iso = _parse_date(dl.get(field) or "")
if iso:
events.append({"title": f"{acronym} — {label}", "start": iso,
"url": url, "color": color, "allDay": True})
start, end = _parse_date_range(dl.get("event_dates") or "")
if start:
ev: dict = {"title": acronym, "start": start, "url": url,
"color": "#457b9d", "allDay": True}
if end:
ev["end"] = end
events.append(ev)
return events
def _calendar_body(venues: dict, icore: dict, deadlines: dict) -> str:
from datetime import datetime
def parse_deadline(s: str):
for fmt in ("%b %d, %Y", "%b %d %Y"):
try:
return datetime.strptime(s.strip(), fmt)
except ValueError:
pass
return datetime.max # TBD / unparseable goes last
rows = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
rank = icore.get(acronym.upper(), "—") or "—"
dl_data = deadlines.get(acronym, {})
deadline = (
dl_data.get("submission_deadline")
or dl_data.get("abstract_deadline")
or "TBD"
)
event = dl_data.get("event_dates") or "—"
rows.append((acronym, deadline, event, rank))
rows.sort(key=lambda r: (parse_deadline(r[1]), r[0]))
lines = [
"\n| Conference | Submission deadline | Event dates | ICORE |",
"|------------|---------------------|-------------|-------|",
]
for acronym, dl, event, rank in rows:
lines.append(f"| [{acronym}](/venues/conferences/{acronym.lower()}/) | {dl} | {event} | {rank} |")
lines += [
"",
"> Deadlines sourced from conference websites and [WikiCFP](http://www.wikicfp.com/).",
"",
]
return "\n".join(lines) + "\n"
def _digest_body(papers: list[dict]) -> str:
lines = [f"\n{len(papers)} papers selected.\n"]
for p in papers:
title = p.get("title") or "Untitled"
authors = p.get("authors") or []
author_str = (
", ".join(authors[:4]) + (" *et al.*" if len(authors) > 4 else "")
if authors else ""
)
tldr = p.get("tldr") or ""
why = p.get("why_notable") or ""
link = p.get("url") or (f"https://doi.org/{p['doi']}" if p.get("doi") else "")
lines += ["---", f"\n### {title}\n"]
if author_str:
lines.append(f"*{author_str}*\n")
if tldr:
lines.append(f"**TL;DR** — {tldr}\n")
if why:
lines.append(f"**Why notable** — {why}\n")
if link:
lines.append(f"[→ Read paper]({link})\n")
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Content generators
# ---------------------------------------------------------------------------
def gen_conferences(venues: dict, icore: dict, deadlines: dict, root: Path) -> int:
count = 0
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "").strip()
if not acronym:
continue
path = root / "venues" / "conferences" / acronym / "_index.md"
existing_fm, _ = read_hugo_file(path)
new_fm: dict = {
"title": acronym,
"full_name": conf.get("full_name", ""),
"domain": conf.get("domain", []),
"type": "conference",
"icore_rank": icore.get(acronym.upper(), ""),
"homepage": conf.get("url", ""),
"dblp_key": conf.get("dblp_key", ""),
"draft": False,
}
dl = deadlines.get(acronym, {})
if dl.get("submission_deadline"):
new_fm["next_deadline"] = dl["submission_deadline"]
elif dl.get("abstract_deadline"):
new_fm["next_deadline"] = dl["abstract_deadline"]
if dl.get("abstract_deadline") and dl.get("submission_deadline"):
new_fm["abstract_deadline"] = dl["abstract_deadline"]
if dl.get("event_dates"):
new_fm["next_event"] = dl["event_dates"]
if dl.get("wikicfp_event_id"):
new_fm["deadline_source"] = f"wikicfp:{dl['wikicfp_event_id']}"
fm = merge_fm(existing_fm, new_fm, PRESERVED_VENUE_FIELDS)
write_hugo_file(path, fm, _conf_body(fm))
count += 1
print(f" {path.relative_to(root.parent)}", file=sys.stderr)
return count
def gen_journals(venues: dict, scimago: dict[str, dict], root: Path) -> int:
count = 0
for journal in venues.get("journals") or []:
acronym = journal.get("acronym", "").strip()
if not acronym:
continue
path = root / "venues" / "journals" / acronym / "_index.md"
existing_fm, _ = read_hugo_file(path)
inline_sjr = {
k: journal[k]
for k in ("scimago_quartile", "scimago_sjr", "scimago_h_index")
if journal.get(k)
}
sjr_data = inline_sjr or scimago_lookup(scimago, journal.get("full_name", ""), journal.get("issn", ""))
new_fm: dict = {
"title": acronym,
"full_name": journal.get("full_name", ""),
"domain": journal.get("domain", []),
"type": "journal",
"issn": journal.get("issn", ""),
"homepage": journal.get("url", ""),
"submission_model": journal.get("submission_model", "rolling"),
"dblp_key": journal.get("dblp_key", ""),
"draft": False,
**sjr_data,
}
fm = merge_fm(existing_fm, new_fm, PRESERVED_VENUE_FIELDS)
write_hugo_file(path, fm, _journal_body(fm))
count += 1
print(f" {path.relative_to(root.parent)}", file=sys.stderr)
return count
def gen_calendar(venues: dict, icore: dict, deadlines: dict, root: Path) -> None:
path = root / "calendar" / "_index.md"
existing_fm, _ = read_hugo_file(path)
new_fm = {
"title": "Submission Deadlines",
"layout": "calendar",
"generated": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"events": _calendar_events(venues, deadlines),
"draft": False,
}
fm = merge_fm(existing_fm, new_fm, set())
write_hugo_file(path, fm, _calendar_body(venues, icore, deadlines))
print(f" {path.relative_to(root.parent)}", file=sys.stderr)
def gen_section_indexes(venues: dict, icore: dict, deadlines: dict, root: Path) -> None:
# venues/_index.md
p = root / "venues" / "_index.md"
fm = merge_fm(read_hugo_file(p)[0], {"title": "Venues", "draft": False}, set())
write_hugo_file(p, fm, _venues_body(venues))
print(f" {p.relative_to(root.parent)}", file=sys.stderr)
# venues/conferences/_index.md
p = root / "venues" / "conferences" / "_index.md"
n = len(venues.get("conferences") or [])
fm = merge_fm(read_hugo_file(p)[0],
{"title": "Conferences", "description": f"{n} tracked conferences", "draft": False}, set())
write_hugo_file(p, fm, _conferences_section_body(venues, icore, deadlines))
print(f" {p.relative_to(root.parent)}", file=sys.stderr)
# venues/journals/_index.md
p = root / "venues" / "journals" / "_index.md"
n = len(venues.get("journals") or [])
fm = merge_fm(read_hugo_file(p)[0],
{"title": "Journals", "description": f"{n} tracked journals", "draft": False}, set())
write_hugo_file(p, fm, _journals_section_body(venues))
print(f" {p.relative_to(root.parent)}", file=sys.stderr)
def gen_digests(papers_dir: Path, root: Path) -> int:
if not papers_dir.exists():
return 0
count = 0
for digest_file in sorted(papers_dir.glob("*-digest.yaml")):
with open(digest_file, encoding="utf-8") as f:
digest = yaml.safe_load(f) or {}
venue = digest.get("venue", "")
year = digest.get("year")
if not venue or not year:
print(f" Skipping {digest_file.name}: missing venue or year", file=sys.stderr)
continue
# Load candidate pool for full metadata (authors, DOI, URL, etc.)
candidates: dict[str, dict] = {}
candidate_file = papers_dir / f"{venue}-{year}-candidates.yaml"
if candidate_file.exists():
with open(candidate_file, encoding="utf-8") as f:
pool = yaml.safe_load(f) or {}
for p in pool.get("papers", []):
key = p.get("dblp_key") or p.get("openalex_id") or p.get("doi") or p.get("title", "")
candidates[key] = p
# Build enriched paper list
papers_out = []
for sel in digest.get("selected", []):
key = sel.get("dblp_key") or sel.get("openalex_id") or sel.get("doi") or sel.get("title", "")
papers_out.append({**candidates.get(key, {}), **sel})
slug = f"{venue}-{year}"
path = root / "digests" / slug / "index.md"
existing_fm, _ = read_hugo_file(path)
new_fm = {
"title": f"{venue} {year} Digest",
"venue": venue,
"year": int(year),
"date": str(digest.get("date", f"{year}-01-01")),
"tags": digest.get("tags", []),
"paper_count": len(papers_out),
"draft": False,
}
fm = merge_fm(existing_fm, new_fm, set())
write_hugo_file(path, fm, _digest_body(papers_out))
count += 1
print(f" {path.relative_to(root.parent)}", file=sys.stderr)
return count
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
ap = argparse.ArgumentParser(description="Generate Hugo content from data files.")
ap.add_argument("--venues", default="site/data/venues.yaml")
ap.add_argument("--icore", default="site/data/rankings/icore.csv")
ap.add_argument("--scimago", default="site/data/rankings/scimago.csv")
ap.add_argument("--deadlines", default="site/data/deadlines.yaml")
ap.add_argument("--papers-dir", default="site/data/papers", dest="papers_dir")
ap.add_argument("--content", default="site/content")
args = ap.parse_args()
venues_path = Path(args.venues)
if not venues_path.exists():
print(f"Error: {venues_path} not found. Populate site/data/venues.yaml first.", file=sys.stderr)
sys.exit(1)
with open(venues_path, encoding="utf-8") as f:
venues = yaml.safe_load(f) or {}
content_root = Path(args.content)
icore = load_icore(args.icore)
scimago = load_scimago(args.scimago)
deadlines = load_deadlines(args.deadlines)
if icore: print(f"Loaded {len(icore)} ICORE entries", file=sys.stderr)
if scimago: print(f"Loaded {len(scimago)} SCImago entries", file=sys.stderr)
if deadlines: print(f"Loaded deadlines for {len(deadlines)} venues", file=sys.stderr)
print("\nGenerating venue section indexes …", file=sys.stderr)
gen_section_indexes(venues, icore, deadlines, content_root)
print("\nGenerating conference pages …", file=sys.stderr)
n_conf = gen_conferences(venues, icore, deadlines, content_root)
print("\nGenerating journal pages …", file=sys.stderr)
n_jour = gen_journals(venues, scimago, content_root)
print("\nGenerating calendar page …", file=sys.stderr)
gen_calendar(venues, icore, deadlines, content_root)
print("\nGenerating digest pages …", file=sys.stderr)
n_dig = gen_digests(Path(args.papers_dir), content_root)
print(f"\nDone: {n_conf} conferences, {n_jour} journals, {n_dig} digests", file=sys.stderr)
if __name__ == "__main__":
main()