Chelsea
6cf35ca034
Implements the parallel.md workstreams (Agents A-E) toward REFACTOR_GOAL.md. Phase 1 — app factory + blueprints + /api/v1: - app.py -> create_app() factory (no module-level app); entrypoints updated - routes/ (auth, pages, settings, admin, assets) + blueprints/api.py at /api/v1 - config.py / extensions.py / security.py extracted; services/ layer added - endpoint names preserved so template url_for() calls keep resolving (static check: all 27 template url_for endpoints are defined routes) Phase 2 — one pluggable filter system: - filter_pipeline/registry.py: @register_stage / @register_plugin + discover_modules - engine._init_stages() instantiates registered stages (no hardcoded dict); process_batch is AI-aware: only short-circuits to the AI-disabled path when a filterset's stages declare requires_ai, so offline filtersets run with AI off - BaseFilterPlugin gets a consumer (stages/plugins.py); Keyword/Quality re-enabled via filter_config.json plugins config - comment tree modes ported to stages/comment_filter.py + shared rules.py; wired into /api/v1/posts/<uuid> and /api/v1/comments/<uuid> via FilterEngine.filter_comments() (fails open) - offline quality_filter filterset exercises plugins+ranker without AI - legacy filter_lib / comment_lib / html_generation_lib / generate_html / active_html path deleted Phase 3 prep — pluggable fetchers + Postgres models: - Post / Comment SQLAlchemy models added to models.py - migrate_content_to_db.py backfill (idempotent by uuid, batched, --dry-run) - platforms/ fetcher registry (extension point) - live reads/writes still go through PostService (disk JSON); cutover deferred Phase 6 — test harness: - pytest.ini + tests/ (conftest with in-memory SQLite fixture, no Postgres; stubbed polling/filter singletons) - test_app_factory.py (route registration, no module-level app), test_api_contracts.py (posts/post_detail/comments/filters shape with monkeypatched post_service + get_filter_engine), test_filter_pipeline.py + test_plugin_contract.py (Flask-free; validated locally 12/12 incl. drop-in stage/plugin discovered with zero core edits) Other: .gitignore added (__pycache__, data/, secrets); pytest in requirements. Verification: py_compile clean across the project; the Flask-free filter-pipeline and plugin-contract tests pass locally. App-factory / API-contract tests need deps+docker to run; runtime flask routes / Auth0-repeated-create_app also gated on docker. Co-Authored-By: Claude <noreply@anthropic.com>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""Shared rule evaluation for posts and comments."""
|
|
|
|
from typing import Any, Dict
|
|
|
|
|
|
def get_nested_value(obj: Dict[str, Any], path: str) -> Any:
|
|
"""Get a nested dict value using dot notation."""
|
|
value = obj
|
|
for key in path.split("."):
|
|
if isinstance(value, dict) and key in value:
|
|
value = value[key]
|
|
else:
|
|
return None
|
|
return value
|
|
|
|
|
|
def evaluate_rule(value: Any, operator: str, target: Any) -> bool:
|
|
"""Evaluate one rule operator."""
|
|
if value is None:
|
|
return False
|
|
|
|
if operator == "equals":
|
|
return value == target
|
|
if operator == "not_equals":
|
|
return value != target
|
|
if operator == "in":
|
|
return value in target
|
|
if operator == "not_in":
|
|
return value not in target
|
|
if operator == "min":
|
|
return value >= target
|
|
if operator == "max":
|
|
return value <= target
|
|
if operator == "after":
|
|
return value > target
|
|
if operator == "before":
|
|
return value < target
|
|
if operator == "contains":
|
|
return target in value
|
|
if operator == "excludes":
|
|
if isinstance(value, list):
|
|
return not any(item in target for item in value)
|
|
return value not in target
|
|
if operator == "includes":
|
|
if isinstance(value, list):
|
|
return target in value
|
|
return False
|
|
if operator == "includes_any":
|
|
if isinstance(value, list) and isinstance(target, list):
|
|
for item in value:
|
|
if isinstance(item, dict):
|
|
for rule in target:
|
|
if (
|
|
isinstance(rule, dict)
|
|
and item.get("topic") == rule.get("topic")
|
|
and item.get("confidence", 0) >= rule.get("confidence_min", 0)
|
|
):
|
|
return True
|
|
elif item in target:
|
|
return True
|
|
return False
|
|
if operator == "min_length":
|
|
return len(str(value)) >= target
|
|
if operator == "max_length":
|
|
return len(str(value)) <= target
|
|
|
|
return False
|
|
|
|
|
|
def apply_rules(item: Dict[str, Any], rules: Dict[str, Dict[str, Any]]) -> bool:
|
|
"""Return True when all field rules pass."""
|
|
if not rules:
|
|
return True
|
|
|
|
for field_path, rule_def in rules.items():
|
|
value = get_nested_value(item, field_path)
|
|
for operator, target in rule_def.items():
|
|
if not evaluate_rule(value, operator, target):
|
|
return False
|
|
return True |