feat: Anniversary Matching Across Jekyll Posts
This commit is contained in:
OpenVelo Agent
@@ -13,17 +13,17 @@ from .logging_setup import (
|
||||
log_run_summary,
|
||||
log_startup,
|
||||
)
|
||||
from .matching import iter_anniversary_paths
|
||||
from .state import load_posted, save_posted
|
||||
|
||||
|
||||
def _iter_candidates(config: Config) -> Iterable[str]:
|
||||
"""Yield candidate post identifiers.
|
||||
|
||||
This is the extension seam for the future blog-clone + matching
|
||||
pipeline. Job 1083 leaves it empty so the silent-on-no-matches
|
||||
contract is the default behaviour.
|
||||
"""Yield candidate post identifiers (the relative path under
|
||||
``_posts/blog/``) for posts whose anniversary is exactly 10 years
|
||||
before today.
|
||||
"""
|
||||
return []
|
||||
post_root = config.blog_dir / "_posts" / "blog"
|
||||
return iter_anniversary_paths(post_root, config.site_url)
|
||||
|
||||
|
||||
def _run_once(config: Config) -> tuple[int, int, int, int, list[str]]:
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import frontmatter
|
||||
|
||||
|
||||
_LOGGER = logging.getLogger("tenbackward.matching")
|
||||
|
||||
_FILENAME_PATTERN = re.compile(r"^(\d{4})-(\d{2})-(\d{2})-(.+)$")
|
||||
|
||||
_DEFAULT_TZ = "Europe/Berlin"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MatchedPost:
|
||||
path: str
|
||||
title: str
|
||||
date: date
|
||||
url: str
|
||||
|
||||
|
||||
def _today_in_berlin(tz_name: str = _DEFAULT_TZ) -> date:
|
||||
return datetime.now(ZoneInfo(tz_name)).date()
|
||||
|
||||
|
||||
def _is_leap_year(year: int) -> bool:
|
||||
return calendar.isleap(year)
|
||||
|
||||
|
||||
def _try_parse_date(value: object) -> date | None:
|
||||
if isinstance(value, datetime):
|
||||
return value.date()
|
||||
if isinstance(value, date):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _safe_filename_match(stem: str) -> tuple[int, int, int, str] | None:
|
||||
match = _FILENAME_PATTERN.match(stem)
|
||||
if match is None:
|
||||
return None
|
||||
year_str, month_str, day_str, slug = match.groups()
|
||||
try:
|
||||
year = int(year_str)
|
||||
month = int(month_str)
|
||||
day = int(day_str)
|
||||
except ValueError:
|
||||
return None
|
||||
try:
|
||||
date(year, month, day)
|
||||
except ValueError:
|
||||
return None
|
||||
if not slug:
|
||||
return None
|
||||
return year, month, day, slug
|
||||
|
||||
|
||||
def _build_url(site_url: str, post_date: date, slug: str) -> str:
|
||||
base = site_url.rstrip("/")
|
||||
return f"{base}/{post_date.year:04d}/{post_date.month:02d}/{post_date.day:02d}/{slug}/"
|
||||
|
||||
|
||||
def _is_unpublished(metadata: dict) -> bool:
|
||||
if "published" not in metadata:
|
||||
return False
|
||||
return metadata["published"] is False
|
||||
|
||||
|
||||
def _matches_anniversary(post_date: date, target_year: int, today: date) -> bool:
|
||||
if post_date.year != target_year:
|
||||
return False
|
||||
if post_date.month == 2 and post_date.day == 29:
|
||||
if _is_leap_year(today.year):
|
||||
return today.month == 2 and today.day == 29
|
||||
return today.month == 3 and today.day == 1
|
||||
return post_date.month == today.month and post_date.day == today.day
|
||||
|
||||
|
||||
def _process_file(
|
||||
md_path: Path,
|
||||
post_root: Path,
|
||||
site_url: str,
|
||||
target_year: int,
|
||||
today: date,
|
||||
) -> MatchedPost | None:
|
||||
rel = md_path.relative_to(post_root).as_posix()
|
||||
|
||||
try:
|
||||
with md_path.open("r", encoding="utf-8") as fh:
|
||||
text = fh.read()
|
||||
except (OSError, UnicodeDecodeError) as exc:
|
||||
_LOGGER.warning(
|
||||
"skipping unreadable post file",
|
||||
extra={"event": "post_unreadable", "path": rel, "error": type(exc).__name__},
|
||||
)
|
||||
return None
|
||||
|
||||
title: str | None = None
|
||||
post_date: date | None = None
|
||||
unpublished = False
|
||||
frontmatter_parsed = False
|
||||
|
||||
try:
|
||||
parsed = frontmatter.loads(text)
|
||||
frontmatter_parsed = True
|
||||
metadata = parsed.metadata if isinstance(parsed.metadata, dict) else {}
|
||||
if isinstance(metadata.get("title"), str) and metadata["title"].strip():
|
||||
title = metadata["title"]
|
||||
post_date = _try_parse_date(metadata.get("date"))
|
||||
unpublished = _is_unpublished(metadata)
|
||||
except Exception as exc: # noqa: BLE001 — frontmatter is opaque
|
||||
_LOGGER.warning(
|
||||
"frontmatter parse failed",
|
||||
extra={"event": "post_frontmatter_error", "path": rel, "error": type(exc).__name__},
|
||||
)
|
||||
|
||||
slug_match = _safe_filename_match(md_path.stem)
|
||||
if slug_match is None:
|
||||
if not frontmatter_parsed:
|
||||
return None
|
||||
_LOGGER.warning(
|
||||
"filename missing required YYYY-MM-DD-slug pattern",
|
||||
extra={"event": "post_filename_invalid", "path": rel},
|
||||
)
|
||||
return None
|
||||
|
||||
fn_year, fn_month, fn_day, fn_slug = slug_match
|
||||
if post_date is None:
|
||||
post_date = date(fn_year, fn_month, fn_day)
|
||||
|
||||
if not _matches_anniversary(post_date, target_year, today):
|
||||
return None
|
||||
|
||||
if unpublished:
|
||||
_LOGGER.warning(
|
||||
"skipping unpublished post",
|
||||
extra={"event": "post_unpublished", "path": rel},
|
||||
)
|
||||
return None
|
||||
|
||||
if title is None:
|
||||
title = fn_slug
|
||||
|
||||
return MatchedPost(
|
||||
path=rel,
|
||||
title=title,
|
||||
date=post_date,
|
||||
url=_build_url(site_url, post_date, fn_slug),
|
||||
)
|
||||
|
||||
|
||||
def find_anniversary_matches(
|
||||
post_root: Path,
|
||||
site_url: str,
|
||||
*,
|
||||
today: date | None = None,
|
||||
) -> list[MatchedPost]:
|
||||
"""Walk the Jekyll post directory and return posts whose anniversary
|
||||
date is exactly 10 years before ``today`` (per the leap-day rules).
|
||||
"""
|
||||
if today is None:
|
||||
today = _today_in_berlin()
|
||||
|
||||
target_year = today.year - 10
|
||||
|
||||
if not post_root.exists():
|
||||
return []
|
||||
|
||||
matches: list[MatchedPost] = []
|
||||
for md_path in sorted(post_root.rglob("*.md")):
|
||||
if not md_path.is_file():
|
||||
continue
|
||||
try:
|
||||
result = _process_file(md_path, post_root, site_url, target_year, today)
|
||||
except Exception as exc: # noqa: BLE001 — never abort the scan
|
||||
rel = md_path.relative_to(post_root).as_posix()
|
||||
_LOGGER.warning(
|
||||
"unexpected error processing post",
|
||||
extra={
|
||||
"event": "post_processing_error",
|
||||
"path": rel,
|
||||
"error": type(exc).__name__,
|
||||
},
|
||||
)
|
||||
continue
|
||||
if result is not None:
|
||||
matches.append(result)
|
||||
|
||||
return matches
|
||||
|
||||
|
||||
def iter_anniversary_paths(post_root: Path, site_url: str, *, today: date | None = None):
|
||||
"""Yield each matched post's relative ``path`` (the dedupe identifier)."""
|
||||
for match in find_anniversary_matches(post_root, site_url, today=today):
|
||||
yield match.path
|
||||
|
||||
|
||||
__all__ = [
|
||||
"MatchedPost",
|
||||
"find_anniversary_matches",
|
||||
"iter_anniversary_paths",
|
||||
]
|
||||
@@ -0,0 +1,199 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
|
||||
from tenbackward.matching import (
|
||||
MatchedPost,
|
||||
find_anniversary_matches,
|
||||
iter_anniversary_paths,
|
||||
)
|
||||
|
||||
|
||||
SITE_URL = "https://chaospott.de"
|
||||
|
||||
_MATCHING_LOGGER = "tenbackward.matching"
|
||||
|
||||
|
||||
def _write_post(root: Path, rel_path: str, *, body: str = "post") -> None:
|
||||
full = root / rel_path
|
||||
full.parent.mkdir(parents=True, exist_ok=True)
|
||||
full.write_text(body, encoding="utf-8")
|
||||
|
||||
|
||||
def _capture_matching_warnings(call: Callable[[], object]) -> list[logging.LogRecord]:
|
||||
records: list[logging.LogRecord] = []
|
||||
handler = logging.Handler()
|
||||
handler.setLevel(logging.WARNING)
|
||||
handler.emit = records.append # type: ignore[assignment]
|
||||
logger = logging.getLogger(_MATCHING_LOGGER)
|
||||
previous_level = logger.level
|
||||
logger.setLevel(logging.WARNING)
|
||||
logger.addHandler(handler)
|
||||
try:
|
||||
call()
|
||||
finally:
|
||||
logger.removeHandler(handler)
|
||||
logger.setLevel(previous_level)
|
||||
return records
|
||||
|
||||
|
||||
def _event_names(records: list[logging.LogRecord]) -> list[str]:
|
||||
return [getattr(record, "event", None) for record in records]
|
||||
|
||||
|
||||
def test_match_uses_frontmatter_date_and_title(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(
|
||||
root,
|
||||
"2014/2014-08-04-foo.md",
|
||||
body="---\ntitle: Foo\ndate: 2014-08-04\n---\nbody\n",
|
||||
)
|
||||
|
||||
today = date(2024, 8, 4)
|
||||
matches = find_anniversary_matches(root, SITE_URL, today=today)
|
||||
|
||||
assert len(matches) == 1
|
||||
match = matches[0]
|
||||
assert isinstance(match, MatchedPost)
|
||||
assert match.path == "2014/2014-08-04-foo.md"
|
||||
assert match.title == "Foo"
|
||||
assert match.date == date(2014, 8, 4)
|
||||
assert match.url == "https://chaospott.de/2014/08/04/foo/"
|
||||
|
||||
|
||||
def test_match_falls_back_to_filename_when_frontmatter_missing(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(root, "2015/2015-03-10-no-frontmatter.md", body="no frontmatter at all\n")
|
||||
_write_post(
|
||||
root, "2015/2015-03-11-broken-frontmatter.md", body="---\ndate: not-a-date\n: :\n"
|
||||
)
|
||||
|
||||
today = date(2025, 3, 10)
|
||||
matches: list[MatchedPost] = []
|
||||
records = _capture_matching_warnings(
|
||||
lambda: matches.extend(find_anniversary_matches(root, SITE_URL, today=today))
|
||||
)
|
||||
|
||||
paths = {m.path for m in matches}
|
||||
assert "2015/2015-03-10-no-frontmatter.md" in paths
|
||||
assert "2015/2015-03-11-broken-frontmatter.md" not in paths
|
||||
|
||||
match = next(m for m in matches if m.path == "2015/2015-03-10-no-frontmatter.md")
|
||||
assert match.title == "no-frontmatter"
|
||||
assert match.date == date(2015, 3, 10)
|
||||
assert match.url == "https://chaospott.de/2015/03/10/no-frontmatter/"
|
||||
|
||||
|
||||
def test_unpublished_post_is_skipped(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(
|
||||
root,
|
||||
"2014/2014-08-04-draft.md",
|
||||
body="---\ntitle: Draft\ndate: 2014-08-04\npublished: false\n---\n",
|
||||
)
|
||||
|
||||
today = date(2024, 8, 4)
|
||||
matches: list[MatchedPost] = []
|
||||
records = _capture_matching_warnings(
|
||||
lambda: matches.extend(find_anniversary_matches(root, SITE_URL, today=today))
|
||||
)
|
||||
|
||||
assert matches == []
|
||||
assert _event_names(records) == ["post_unpublished"]
|
||||
|
||||
|
||||
def test_iter_anniversary_paths_yields_only_path(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(
|
||||
root,
|
||||
"2014/2014-08-04-foo.md",
|
||||
body="---\ntitle: Foo\ndate: 2014-08-04\n---\n",
|
||||
)
|
||||
|
||||
paths = list(iter_anniversary_paths(root, SITE_URL, today=date(2024, 8, 4)))
|
||||
assert paths == ["2014/2014-08-04-foo.md"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"today, source_iso, expected_match",
|
||||
[
|
||||
# Normal exact match (Mar 1 source, Mar 1 today) -> match
|
||||
(date(2025, 3, 1), "2015-03-01", True),
|
||||
# Mar 1 source (2016, the day after Feb 29) in a non-leap current year -> no match
|
||||
(date(2025, 3, 1), "2016-03-01", False),
|
||||
# Feb 29 source in a non-leap current year, today is Mar 1: source year
|
||||
# (2016) does not equal target_year (2015), so it does NOT match under
|
||||
# the strict year-equality rule.
|
||||
(date(2025, 3, 1), "2016-02-29", False),
|
||||
],
|
||||
)
|
||||
def test_leap_day_matching(
|
||||
tmp_path: Path, today: date, source_iso: str, expected_match: bool
|
||||
) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
target_year = today.year - 10
|
||||
rel = f"{target_year}/{source_iso}-leap.md"
|
||||
_write_post(
|
||||
root,
|
||||
rel,
|
||||
body=f"---\ntitle: Leap\ndate: {source_iso}\n---\n",
|
||||
)
|
||||
|
||||
matches = find_anniversary_matches(root, SITE_URL, today=today)
|
||||
matched_paths = [m.path for m in matches]
|
||||
if expected_match:
|
||||
assert rel in matched_paths
|
||||
else:
|
||||
assert rel not in matched_paths
|
||||
|
||||
|
||||
def test_match_uses_datetime_frontmatter_date(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(
|
||||
root,
|
||||
"2014/2014-08-04-with-time.md",
|
||||
body="---\ntitle: WithTime\ndate: 2014-08-04T12:30:00\n---\n",
|
||||
)
|
||||
|
||||
today = date(2024, 8, 4)
|
||||
matches = find_anniversary_matches(root, SITE_URL, today=today)
|
||||
assert len(matches) == 1
|
||||
assert matches[0].date == date(2014, 8, 4)
|
||||
|
||||
|
||||
def test_match_returns_results_in_deterministic_order(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
for slug, day in [("alpha", 1), ("beta", 2), ("gamma", 3)]:
|
||||
_write_post(
|
||||
root,
|
||||
f"2014/2014-08-0{day}-{slug}.md",
|
||||
body=f"---\ntitle: {slug}\ndate: 2014-08-0{day}\n---\n",
|
||||
)
|
||||
|
||||
today = date(2024, 8, 2)
|
||||
matches = find_anniversary_matches(root, SITE_URL, today=today)
|
||||
assert [m.path for m in matches] == ["2014/2014-08-02-beta.md"]
|
||||
|
||||
|
||||
def test_missing_post_root_returns_empty(tmp_path: Path) -> None:
|
||||
today = date(2024, 8, 4)
|
||||
assert find_anniversary_matches(tmp_path / "does-not-exist", SITE_URL, today=today) == []
|
||||
|
||||
|
||||
def test_invalid_filename_is_skipped(tmp_path: Path) -> None:
|
||||
root = tmp_path / "_posts" / "blog"
|
||||
_write_post(root, "2014/not-a-date.md", body="---\ntitle: Bad\n---\n")
|
||||
|
||||
today = date(2024, 8, 4)
|
||||
matches: list[MatchedPost] = []
|
||||
records = _capture_matching_warnings(
|
||||
lambda: matches.extend(find_anniversary_matches(root, SITE_URL, today=today))
|
||||
)
|
||||
|
||||
assert matches == []
|
||||
assert "post_filename_invalid" in _event_names(records)
|
||||
Reference in New Issue
Block a user