Files
publish-assistant/src/publish_assistant/generate_content.py
khannurien ee78e30d17
All checks were successful
Build and deploy static pages / build-and-push (push) Successful in 14s
Add an embedded systems topic
Eight conferences and six journals, curated for the computer-science side of
the field. EDA and circuit-design venues are deliberately excluded — DAC, DATE,
ICCAD and CODES+ISSS are canonical for embedded systems but not for the kind of
work this site tracks — and venues.yaml carries that rationale as a header
comment so the next person doesn't "helpfully" add them back.

Two curation calls worth recording. SenSys, IPSN and IoTDI merged in 2026 into
one conference, the ACM/IEEE International Conference on Embedded Artificial
Intelligence and Sensing Systems; it is tracked under the SenSys acronym, which
is what ICORE and DBLP still use, and IPSN and IoTDI are not listed separately.
RTCSA and ICCPS were dropped — the first is CORE B with no announced 2027
edition, the second carries no CORE rank at all.

Deadlines are all source: manual, verified against the official CFPs. Only EWSN
has a WikiCFP entry for its current edition; every other series there stops at
an edition that has already been held, so the rest are wikicfp_id: false. ECRTS
and LCTES have no announced next edition and sit under missing:, which renders
as TBD. With the stale-deadline handling from 5b06f48 this reads correctly:
only RTAS shows an open deadline, and the rest land in the elapsed-rounds table
rather than advertising passed dates as upcoming.

The topic also exposed three generator bugs that cloud-edge never hit:

_journals_section_body() read scimago_* straight off the venues.yaml entry, so
the journals index showed a dash for every journal without inline values, while
the journal pages resolved the same numbers from rankings/scimago.csv. Both now
share a _journal_ranks() helper, and gen_section_indexes() takes the SCImago
table to feed it. cloud-edge never saw this because all seven of its journals
carry inline values.

The SCImago CSV writes SJR with a decimal comma (0,864). Nothing consumed those
values before, so the loader now normalises them to a dot and CSV-derived
numbers render like the inline ones.

gen_digests() now writes digests/_index.md. Hugo only auto-creates a section
page when the section has a listable page, and a topic with no digests yet
holds nothing but build.list: never stubs, so the Digests nav link 404'd. It
preserves any existing body, leaving cloud-edge's file unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-18 13:33:53 +00:00

994 lines
42 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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/<topic>/venues.yaml master venue list
site/data/rankings/icore.csv ICORE conference ranks (shared)
site/data/rankings/scimago.csv SCImago journal ranks (shared, optional)
site/data/<topic>/deadlines.yaml deadline data
site/data/<topic>/papers/<V>-<Y>-candidates.yaml paper pools
site/data/<topic>/papers/<V>-<Y>-digest.yaml agent-curated selections
Writes:
site/content/<topic>/venues/conferences/<acronym>/_index.md
site/content/<topic>/venues/journals/<acronym>/_index.md
site/content/<topic>/calendar/_index.md
site/content/<topic>/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
from publish_assistant.cycles import (
cycle_deadline,
next_cycle,
normalise_cycles,
open_cycles,
parse_date,
)
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
# - "cycles" / "next_cycle" rebuilt from deadlines.yaml every run, so a venue
# dropping back to a single round does not keep a stale cycle list
_STRIP_FROM_EXISTING = {"url", "papers", "cycles", "next_cycle"}
# ---------------------------------------------------------------------------
# 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().replace(",", "."),
"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 {}
deadlines = data.get("deadlines") or {}
for entry in deadlines.values():
if entry:
normalise_cycles(entry)
return deadlines
def load_venue_digests(papers_dir: Path) -> dict[str, list[dict]]:
"""Return {acronym: [digest_info_dict, ...]} sorted newest-first."""
result: dict[str, list[dict]] = {}
if not papers_dir.exists():
return result
for f in sorted(papers_dir.glob("*-digest.yaml"), reverse=True):
with open(f, encoding="utf-8") as fp:
d = yaml.safe_load(fp) or {}
venue = d.get("venue", "")
year = d.get("year")
if venue and year:
result.setdefault(venue, []).append({
"year": int(year),
"slug": f"{venue}-{year}",
"paper_count": len(d.get("selected") or d.get("papers") or []),
"tags": d.get("tags", []),
"date": str(d.get("date", f"{year}-01-01")),
})
return result
# ---------------------------------------------------------------------------
# Markdown body renderers
# ---------------------------------------------------------------------------
def _bold_rank(rank: str) -> str:
"""Bold a rank that may carry a literal asterisk. '**A***' closes the
emphasis on the second star, leaving the third one loose outside it."""
return "**" + str(rank).replace("*", r"\*") + "**"
def _conf_body(fm: dict, digests: list[dict] | None = None, base_path: str = "") -> 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 ""
notification = fm.get("notification") or ""
camera_ready = fm.get("camera_ready") or ""
event = fm.get("next_event") or ""
location = fm.get("location") or ""
cfp_url = fm.get("cfp_url") or ""
domains = ", ".join(f"`{d}`" for d in (fm.get("domain") or []))
notes = fm.get("notes") or ""
src = fm.get("deadline_source") or ""
title = fm.get("title", "")
lines = [f"\n*{fm.get('full_name', '')}*\n"]
lines += [
"| | |", "|---|---|",
f"| **ICORE Rank** | {_bold_rank(rank)} |",
f"| **Domains** | {domains} |",
]
if hp:
lines.append(f"| **Website** | [{hp}]({hp}) |")
if digests:
latest = sorted(digests, key=lambda x: x["year"], reverse=True)[0]
slug = latest["slug"].lower()
year = latest["year"]
count = latest.get("paper_count", 0)
suffix = f"{count} papers" if count else "*(digest pending)*"
lines.append(f"| **Latest digest** | [{title} {year}]({base_path}/digests/{slug}/) {suffix} |")
cycles = fm.get("cycles") or []
if len(cycles) > 1:
lines += [
"\n## Submission Cycles\n",
"| Cycle | Abstract | Paper deadline | Notification | Camera-ready |",
"|---|---|---|---|---|",
]
next_name = fm.get("next_cycle")
marked = False
for cycle in cycles:
name = cycle.get("name") or ""
if not marked and name == next_name:
name = f"**{name}** (next)"
marked = True
paper = cycle.get("submission_deadline") or cycle.get("abstract_deadline") or "TBD"
abstract = cycle.get("abstract_deadline") or ""
abs_cell = abstract if (abstract and abstract != paper) else ""
lines.append(
f"| {name} | {abs_cell} | {paper} | {cycle.get('notification') or ''} "
f"| {cycle.get('camera_ready') or ''} |"
)
lines += ["\n| | |", "|---|---|"]
else:
lines += [
"\n## Upcoming Deadline\n",
"| | |", "|---|---|",
]
if abstract_dl and abstract_dl != dl:
lines.append(f"| **Abstract deadline** | {abstract_dl} |")
lines.append(f"| **Paper deadline** | {dl} |")
if notification:
lines.append(f"| **Notification** | {notification} |")
if camera_ready:
lines.append(f"| **Camera-ready** | {camera_ready} |")
lines.append(f"| **Event dates** | {event} |")
if location:
lines.append(f"| **Location** | {location} |")
if cfp_url:
lines.append(f"| **CFP** | [{cfp_url}]({cfp_url}) |")
if src:
lines.append(f"| **Source** | `{src}` |")
if digests:
lines += ["\n## Previous Editions\n"]
for d in sorted(digests, key=lambda x: x["year"], reverse=True):
slug = d["slug"].lower()
year = d["year"]
count = d.get("paper_count", 0)
tags = d.get("tags") or []
lines.append(f"### {title} {year}\n")
lines += ["| | |", "|---|---|"]
if count:
lines.append(f"| **Papers digested** | {count} |")
if tags:
lines.append(f"| **Topics** | {', '.join(f'`{t}`' for t in tags)} |")
lines.append(f"| **Digest** | [→ {title} {year} Digest]({base_path}/digests/{slug}/) |")
lines.append("")
if notes:
lines += ["\n## Notes\n", notes]
return "\n".join(lines) + "\n"
def _journal_body(fm: dict, digests: list[dict] | None = None, base_path: str = "",
special_issues: list[dict] | None = None) -> 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 ""
title = fm.get("title", "")
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 digests:
latest = sorted(digests, key=lambda x: x["year"], reverse=True)[0]
slug = latest["slug"].lower()
year = latest["year"]
count = latest.get("paper_count", 0)
suffix = f"{count} papers" if count else "*(digest pending)*"
lines.append(f"| **Latest digest** | [{title} {year}]({base_path}/digests/{slug}/) {suffix} |")
if special_issues:
lines += ["\n## Special Issues\n"]
lines += [
"| Title | Abstract | Paper deadline | Notification | Guest Editors |",
"|---|---|---|---|---|",
]
for si in special_issues:
si_title = si.get("title", "Special Issue")
abstract = si.get("abstract_deadline") or ""
paper = si.get("submission_deadline") or si.get("abstract_deadline") or "TBD"
notif = si.get("notification") or ""
editors = ", ".join(si.get("guest_editors", [])) or ""
abs_cell = abstract if (abstract and abstract != paper) else ""
cfp_url = si.get("cfp_url", "")
if cfp_url:
si_title = f"[{si_title}]({cfp_url})"
lines.append(f"| {si_title} | {abs_cell} | {paper} | {notif} | {editors} |")
lines.append("")
if digests:
lines += ["\n## Previous Editions\n"]
for d in sorted(digests, key=lambda x: x["year"], reverse=True):
slug = d["slug"].lower()
year = d["year"]
count = d.get("paper_count", 0)
tags = d.get("tags") or []
lines.append(f"### {title} {year}\n")
lines += ["| | |", "|---|---|"]
if count:
lines.append(f"| **Papers digested** | {count} |")
if tags:
lines.append(f"| **Topics** | {', '.join(f'`{t}`' for t in tags)} |")
lines.append(f"| **Digest** | [→ {title} {year} Digest]({base_path}/digests/{slug}/) |")
lines.append("")
if notes:
lines += ["\n## Notes\n", notes]
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Date helpers
# ---------------------------------------------------------------------------
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, base_path: str = "") -> 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"
f"- **[Conferences →]({base_path}/venues/conferences/)** ranked by ICORE (A*, A, B, C)\n"
f"- **[Journals →]({base_path}/venues/journals/)** ranked by SCImago quartile (Q1Q4)\n"
)
def _conferences_section_body(venues: dict, icore: dict, deadlines: dict,
venue_digests: dict | None = None, base_path: str = "") -> str:
rows = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
rank = icore.get(acronym.upper(), "") or ""
dl = deadlines.get(acronym, {})
# The venue index lists venues, not cycles: show the next open round and
# name it, so a passed round never reads as "you missed this conference".
nxt = next_cycle(dl)
abstract = nxt.get("abstract_deadline") or ""
paper = nxt.get("submission_deadline") or nxt.get("abstract_deadline") or "TBD"
if len(dl.get("cycles") or []) > 1 and nxt.get("name"):
paper = f"{paper} ({nxt['name']})"
notif = nxt.get("notification") or ""
event = dl.get("event_dates") or ""
location = dl.get("location") or ""
domains = ", ".join(f"`{d}`" for d in conf.get("domain", []))
n_dig = len((venue_digests or {}).get(acronym, []))
rows.append((_ICORE_ORDER.get(rank, 99), acronym, conf.get("full_name", ""),
rank, abstract, paper, notif, event, location, domains, n_dig))
rows.sort(key=lambda r: (r[0], r[1]))
lines = [
"\n| Conference | Full name | ICORE | Domains | Abstract | Paper deadline | Notification | Event | Location | Digests |",
"|---|---|---|---|---|---|---|---|---|---|",
]
for _, acronym, full_name, rank, abstract, paper, notif, event, location, domains, n_dig in rows:
link = f"[{acronym}]({base_path}/venues/conferences/{acronym.lower()}/)"
abs_cell = abstract if (abstract and abstract != paper) else ""
dig_cell = str(n_dig) if n_dig else ""
lines.append(f"| {link} | {full_name} | {_bold_rank(rank)} | {domains} | {abs_cell} | {paper} | {notif} | {event} | {location} | {dig_cell} |")
return "\n".join(lines) + "\n"
def _journal_ranks(journal: dict, scimago: dict[str, dict]) -> dict:
"""SCImago values for one journal: inline `scimago_*` fields win, CSV fills the gap."""
inline = {
k: journal[k]
for k in ("scimago_quartile", "scimago_sjr", "scimago_h_index")
if journal.get(k)
}
return inline or scimago_lookup(scimago, journal.get("full_name", ""), journal.get("issn", ""))
def _journals_section_body(venues: dict, scimago: dict[str, dict] | None = None,
venue_digests: dict | None = None, base_path: str = "") -> str:
rows = []
for j in venues.get("journals") or []:
acronym = j.get("acronym", "")
ranks = _journal_ranks(j, scimago or {})
q = ranks.get("scimago_quartile") or ""
sjr = ranks.get("scimago_sjr") or ""
h = ranks.get("scimago_h_index") or ""
model = (j.get("submission_model") or "rolling").title()
domains = ", ".join(f"`{d}`" for d in j.get("domain", []))
n_dig = len((venue_digests or {}).get(acronym, []))
rows.append((_QUARTILE_ORDER.get(q, 99), acronym, j.get("full_name", ""), q, sjr, h, model, domains, n_dig))
rows.sort(key=lambda r: (r[0], r[1]))
lines = [
"\n| Journal | Full name | SCImago | SJR | H-Index | Model | Domains | Digests |",
"|---|---|---|---|---|---|---|---|",
]
for _, acronym, full_name, q, sjr, h, model, domains, n_dig in rows:
link = f"[{acronym}]({base_path}/venues/journals/{acronym.lower()}/)"
dig_cell = str(n_dig) if n_dig else ""
lines.append(f"| {link} | {full_name} | **{q}** | {sjr} | {h} | {model} | {domains} | {dig_cell} |")
return "\n".join(lines) + "\n"
# ---------------------------------------------------------------------------
# Calendar events builder (for FullCalendar)
# ---------------------------------------------------------------------------
# One chip colour per event kind. Backgrounds stay pale and dark ink is set
# explicitly, so a chip is legible on both the light and the dark site theme —
# FullCalendar's default white event text would not be.
EVENT_COLORS = {
"abstract": ("#f0d2a4", "#4a3212"),
"paper": ("#f2b3ac", "#5a1f19"),
"notification": ("#b9dfd0", "#14402f"),
"event": ("#bdd2ee", "#1a3557"),
"special": ("#d8c8f0", "#35215c"),
}
def _calendar_events(venues: dict, deadlines: dict, base_path: str = "") -> list[dict]:
events: list[dict] = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
dl = deadlines.get(acronym, {})
url = f"{base_path}/venues/conferences/{acronym.lower()}/"
location = dl.get("location") or ""
for cycle in dl.get("cycles") or []:
cycle_name = cycle.get("name") or ""
prefix = f"{acronym} {cycle_name}".strip()
cfp_url = cycle.get("cfp_url") or dl.get("cfp_url") or ""
for field, label, kind in [
("abstract_deadline", "abstract deadline", "abstract"),
("submission_deadline", "paper deadline", "paper"),
("notification", "notification", "notification"),
]:
color, text_color = EVENT_COLORS[kind]
iso = parse_date(cycle.get(field) or "")
if iso:
ev: dict = {"title": f"{prefix}{label}", "start": iso,
"url": url, "color": color,
"textColor": text_color, "allDay": True}
props: dict = {}
if location:
props["location"] = location
if cfp_url:
props["cfp_url"] = cfp_url
if cycle_name:
props["cycle"] = cycle_name
if props:
ev["extendedProps"] = props
events.append(ev)
start, end = _parse_date_range(dl.get("event_dates") or "")
if start:
color, text_color = EVENT_COLORS["event"]
ev = {"title": acronym, "start": start, "url": url,
"color": color, "textColor": text_color, "allDay": True}
if end:
ev["end"] = end
if location:
ev["extendedProps"] = {"location": location}
events.append(ev)
# Journal special issue deadlines — one violet series, distinct from conferences
journal_color, journal_text_color = EVENT_COLORS["special"]
for journal in venues.get("journals") or []:
acronym = journal.get("acronym", "")
url = f"{base_path}/venues/journals/{acronym.lower()}/"
for si in journal.get("special_issues") or []:
si_title = si.get("title", "Special Issue")
cfp_url = si.get("cfp_url", "")
for field, label in [
("abstract_deadline", "abstract deadline"),
("submission_deadline", "paper deadline"),
("notification", "notification"),
]:
# Notifications are often month-only ('Jan 2027'); deadlines are not.
iso = parse_date(si.get(field) or "", month_only=(field == "notification"))
if iso:
ev = {"title": f"{acronym} SI — {si_title}{label}",
"start": iso, "url": url, "color": journal_color,
"textColor": journal_text_color, "allDay": True}
if cfp_url:
ev["extendedProps"] = {"cfp_url": cfp_url}
events.append(ev)
return events
def _calendar_body(venues: dict, icore: dict, deadlines: dict, base_path: str = "") -> str:
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
lines: list[str] = []
conf_rows = []
stale_rows = []
for conf in venues.get("conferences") or []:
acronym = conf.get("acronym", "")
rank = icore.get(acronym.upper(), "") or ""
dl_data = deadlines.get(acronym, {})
event = dl_data.get("event_dates") or ""
location = dl_data.get("location") or ""
# One row per still-open submission cycle — a venue's later rounds are
# the point, but an elapsed round is not something a reader can act on,
# and sorting by deadline parks it at the top of the table.
still_open = open_cycles(dl_data)
if not still_open:
# Every recorded round has passed: the venue is still tracked, its
# next CFP just isn't out (or isn't fetched) yet. Listed separately
# below so a stale date never reads as an upcoming deadline.
last = next_cycle(dl_data)
stale_rows.append((acronym, conf.get("full_name", ""),
cycle_deadline(last) or "TBD", event, location, rank))
continue
for cycle in still_open:
abstract = cycle.get("abstract_deadline") or ""
paper = cycle.get("submission_deadline") or cycle.get("abstract_deadline") or "TBD"
notif = cycle.get("notification") or ""
conf_rows.append((acronym, conf.get("full_name", ""), cycle.get("name") or "",
abstract, paper, notif, event, location, rank))
conf_rows.sort(key=lambda r: (parse_deadline(r[4]), r[0]))
stale_rows.sort(key=lambda r: (parse_deadline(r[2]), r[0]))
lines.append("\n## Conference Deadlines\n")
if conf_rows:
lines += [
"| Venue | Name | Cycle | Abstract | Paper deadline | Notification | Event dates | Location | ICORE |",
"|---|---|---|---|---|---|---|---|---|",
]
for acronym, full_name, cycle_name, abstract, paper, notif, event, location, rank in conf_rows:
abs_cell = abstract if (abstract and abstract != paper) else ""
link = f"[{acronym}]({base_path}/venues/conferences/{acronym.lower()}/)"
lines.append(
f"| {link} | {full_name} | {cycle_name} | {abs_cell} | {paper} | {notif} "
f"| {event} | {location} | {rank} |"
)
else:
lines.append("No open call at any tracked conference.")
if stale_rows:
lines += [
"\n### Awaiting the next call\n",
"Tracked venues whose recorded rounds have all elapsed — the next CFP "
"has not been published or picked up yet. Dates below are the last "
"round that ran, kept for reference only.\n",
"| Venue | Name | Last recorded deadline | Event dates | Location | ICORE |",
"|---|---|---|---|---|---|",
]
for acronym, full_name, last_deadline, event, location, rank in stale_rows:
link = f"[{acronym}]({base_path}/venues/conferences/{acronym.lower()}/)"
lines.append(
f"| {link} | {full_name} | {last_deadline} | {event} | {location} | {rank} |"
)
si_rows = []
for journal in venues.get("journals") or []:
acronym = journal.get("acronym", "")
for si in journal.get("special_issues") or []:
abstract = si.get("abstract_deadline") or ""
paper = si.get("submission_deadline") or si.get("abstract_deadline") or "TBD"
notif = si.get("notification") or ""
si_title = si.get("title", "Special Issue")
cfp_url = si.get("cfp_url", "")
if cfp_url:
si_title = f"[{si_title}]({cfp_url})"
editors = ", ".join(si.get("guest_editors", [])) or ""
si_rows.append((acronym, si_title, abstract, paper, notif, editors))
si_rows.sort(key=lambda r: (parse_deadline(r[3]), r[0]))
# Always render the section: dropping it when empty made the page read as if
# journals weren't tracked at all, rather than as having no open call.
lines.append("\n## Journal Special Issues\n")
if si_rows:
lines += [
"| Venue | Title | Abstract | Paper deadline | Notification | Guest Editors |",
"|---|---|---|---|---|---|",
]
for acronym, si_title, abstract, paper, notif, editors in si_rows:
abs_cell = abstract if (abstract and abstract != paper) else ""
link = f"[{acronym}]({base_path}/venues/journals/{acronym.lower()}/)"
lines.append(
f"| {link} | {si_title} | {abs_cell} | {paper} | {notif} | {editors} |"
)
else:
lines.append(
"No open call at any tracked journal. These journals use rolling "
"submission, so special issue CFPs are their only dated deadlines — "
"and none are recorded right now."
)
lines += [
"",
"> Deadlines sourced from conference websites, [WikiCFP](http://www.wikicfp.com/), "
"and journal special issue CFPs.",
"",
]
return "\n".join(lines) + "\n"
def _digest_body(papers: list[dict]) -> str:
if not papers:
return "\nNo papers selected yet.\n"
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,
venue_digests: dict[str, list[dict]] | None = None,
base_path: str = "") -> 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, {})
cycles = dl.get("cycles") or []
nxt = next_cycle(dl)
if nxt.get("submission_deadline"):
new_fm["next_deadline"] = nxt["submission_deadline"]
elif nxt.get("abstract_deadline"):
new_fm["next_deadline"] = nxt["abstract_deadline"]
if nxt.get("abstract_deadline") and nxt.get("submission_deadline"):
new_fm["abstract_deadline"] = nxt["abstract_deadline"]
if nxt.get("notification"):
new_fm["notification"] = nxt["notification"]
if nxt.get("camera_ready"):
new_fm["camera_ready"] = nxt["camera_ready"]
# Only multi-cycle venues carry the full list; single-cycle pages keep
# the flat front matter they have always had.
if len(cycles) > 1:
new_fm["cycles"] = cycles
if nxt.get("name"):
new_fm["next_cycle"] = nxt["name"]
if dl.get("event_dates"):
new_fm["next_event"] = dl["event_dates"]
if dl.get("location"):
new_fm["location"] = dl["location"]
if dl.get("cfp_url"):
new_fm["cfp_url"] = dl["cfp_url"]
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)
digests = (venue_digests or {}).get(acronym)
write_hugo_file(path, fm, _conf_body(fm, digests, base_path))
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,
venue_digests: dict[str, list[dict]] | None = None,
base_path: str = "") -> 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)
sjr_data = _journal_ranks(journal, scimago)
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)
digests = (venue_digests or {}).get(acronym)
write_hugo_file(path, fm, _journal_body(fm, digests, base_path,
journal.get("special_issues")))
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,
base_path: str = "") -> 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, base_path),
"draft": False,
}
fm = merge_fm(existing_fm, new_fm, set())
write_hugo_file(path, fm, _calendar_body(venues, icore, deadlines, base_path))
print(f" {path.relative_to(root.parent)}", file=sys.stderr)
def gen_section_indexes(venues: dict, icore: dict, scimago: dict[str, dict],
deadlines: dict, root: Path,
venue_digests: dict | None = None, base_path: str = "") -> 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, base_path))
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, venue_digests, base_path))
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, scimago, venue_digests, base_path))
print(f" {p.relative_to(root.parent)}", file=sys.stderr)
def _upcoming_year(acronym: str, deadlines: dict) -> int:
"""Extract the year of the upcoming event from deadlines, falling back to current year."""
event_dates = (deadlines.get(acronym) or {}).get("event_dates") or ""
m = re.search(r'\b(20\d\d)\b', event_dates)
return int(m.group(1)) if m else datetime.now().year
def gen_digests(papers_dir: Path, root: Path,
venues: dict | None = None, deadlines: dict | None = None) -> int:
real_slugs: set[str] = set()
count = 0
# Section index. Hugo only auto-creates one when the section has a listable
# page, and a fresh topic holds nothing but `build.list: never` stubs — so
# without this file the Digests nav link 404s until the first real digest.
index_path = root / "digests" / "_index.md"
index_fm, index_body = read_hugo_file(index_path)
index_fm = merge_fm(index_fm, {"title": "Digests", "layout": "digests", "draft": False}, set())
write_hugo_file(index_path, index_fm,
f"\n{index_body.strip() or 'tl;drs for the top papers.'}\n")
print(f" {index_path.relative_to(root.parent)}", file=sys.stderr)
if papers_dir.exists():
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
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
items = digest.get("selected") or []
if not items:
for p in digest.get("papers") or []:
items.append({
"title": p.get("title", ""),
"authors": p.get("authors", []),
"tldr": p.get("reason", ""),
"why_notable": "",
})
papers_out = []
for sel in items:
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}"
real_slugs.add(slug)
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)
# Generate stubs for every venue that has no digest for its upcoming edition
if venues and deadlines is not None:
all_acronyms = (
[c.get("acronym", "") for c in (venues.get("conferences") or [])] +
[j.get("acronym", "") for j in (venues.get("journals") or [])]
)
for acronym in all_acronyms:
if not acronym:
continue
year = _upcoming_year(acronym, deadlines)
slug = f"{acronym}-{year}"
if slug in real_slugs:
continue
path = root / "digests" / slug / "index.md"
if path.exists():
continue # never overwrite existing stubs (may have been customised)
new_fm = {
"title": f"{acronym} {year} Digest",
"venue": acronym,
"year": year,
"date": f"{year}-01-01",
"tags": [],
"paper_count": 0,
"draft": False,
"build": {"list": "never"},
}
write_hugo_file(path, new_fm, _digest_body([]))
count += 1
print(f" {path.relative_to(root.parent)} (stub)", 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/cloud-edge/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/cloud-edge/deadlines.yaml")
ap.add_argument("--papers-dir", default="site/data/cloud-edge/papers", dest="papers_dir")
ap.add_argument("--content", default="site/content/cloud-edge")
ap.add_argument("--base-path", default="/cloud-edge", dest="base_path",
help="URL prefix for all internal links, e.g. '/cloud-edge'")
args = ap.parse_args()
venues_path = Path(args.venues)
if not venues_path.exists():
print(f"Error: {venues_path} not found.", file=sys.stderr)
sys.exit(1)
with open(venues_path, encoding="utf-8") as f:
venues = yaml.safe_load(f) or {}
base_path = args.base_path.rstrip("/")
content_root = Path(args.content)
papers_dir = Path(args.papers_dir)
icore = load_icore(args.icore)
scimago = load_scimago(args.scimago)
deadlines = load_deadlines(args.deadlines)
venue_digests = load_venue_digests(papers_dir)
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, scimago, deadlines, content_root, venue_digests, base_path)
print("\nGenerating conference pages …", file=sys.stderr)
n_conf = gen_conferences(venues, icore, deadlines, content_root, venue_digests, base_path)
print("\nGenerating journal pages …", file=sys.stderr)
n_jour = gen_journals(venues, scimago, content_root, venue_digests, base_path)
print("\nGenerating calendar page …", file=sys.stderr)
gen_calendar(venues, icore, deadlines, content_root, base_path)
print("\nGenerating digest pages …", file=sys.stderr)
n_dig = gen_digests(papers_dir, content_root, venues, deadlines)
print(f"\nDone: {n_conf} conferences, {n_jour} journals, {n_dig} digests", file=sys.stderr)
if __name__ == "__main__":
main()