Tech Spec — Django subcommand to enable statuses by neighborhood constraints

Tailored to the provided models: Landing_Page, O_Search, Document.

Goal

Success depends on high-coverage unit tests and adherence to clean-code practices.

Please send me your bitbucket username to share with you the initial code

Compute and set the status flags on O_Search rows (e.g., status_direct, status_sublet, status_coworking, status_nearby, plus cluster/industry/zipcode variants) based on neighborhood/submarket/zipcode constraints derived from Landing_Page and field values within O_Search. Also (optionally) maintain the M2M page associations (pages, pages2, pages3, pages4) to reflect eligibility.

Relevant models (given)

Status semantics

• status_direct: Eligible for a specific Landing Page that exactly matches neighborhood/submarket & compatible type_of_space/space_use.
• status_sublet: Eligible on sublet pages when space_type or business rule marks as sublet.
• status_coworking: Eligible for coworking pages when space_type == "coworking" (or configured mapping).
• status_nearby: Eligible on “nearby” Landing Pages (from nearby_landing_page_ids) when no exact neighborhood match or as an additional placement within radius / logical adjacency.
• cluster_status_* / industry_status_* / zipcode_status_*: rollups set to 1 if any corresponding granular status is 1 inside that dimension.

Constraints & matching

1. Landing Page activation: only pages with state=1 are considered.
2. Neighborhood match: case-insensitive equality between O_Search.submarket_name or O_Search.submarket_cluster and Landing_Page.neighborhood_name or submarket_cluster (prefer exact neighborhood_name, fallback to cluster).
3. Zipcode match: if Landing_Page.zipcode is set, require int(O_Search.zipcode) equality; treat empty/invalid zipcode in O_Search as non-match.
4. Type-of-space: require compatibility between Landing_Page.type_of_space and O_Search.space_type (mapping table or simple equals, configurable).
5. Language: if Landing_Page.language is set, prefer or force listings with the same language context (configurable: soft vs hard).
6. Nearby: when no exact neighborhood page matches, or when configured to also surface adjacent inventory, use nearby_landing_page_ids (comma list) to set status_nearby=1 and link M2M.

Business rules

1. Direct: If an active Landing_Page matches neighborhood/cluster (+ optional zipcode) and type_of_space is compatible → status_direct=1; else 0.
2. Sublet: If O_Search.space_type or space_use indicates sublet (e.g., contains "sublet") and page compatible → status_sublet=1.
3. Coworking: If space_type == "coworking" and page compatible → status_coworking=1.
4. Nearby: If no direct match, or adjacency policy enabled, propagate to pages listed in nearby_landing_page_ids → status_nearby=1.
5. Cluster / Industry / Zip rollups: set e.g. cluster_status_direct = 1 if any direct match within the same submarket_cluster; similarly for industry_status_* (by space_use) and zipcode_status_* (by zipcode).
6. Manhattan flag (example): if submarket_cluster contains "Manhattan" or zipcode in configured set → manhattan_status=1 else 0.

CLI interface

python manage.py recompute_statuses \
  [--ids 1,2,3]                         # target O_Search ids
  [--filter "submarket_cluster=Midtown,status=0"]  # safe subset k=v pairs
  [--since "2025-09-01T00:00:00"]       # updated_at >= since
  [--chunk-size 1000]                   # default 500
  [--dry-run]                           # no writes
  [--update-m2m]                        # also refresh M2M page links
  [--language en]                       # scope by Landing_Page.language
  [--verbose]

Algorithm

1. Resolve candidate Landing Pages:
   • LP = Landing_Page.objects.filter(state=1) (+ optional language filter).
   • Pre-index LPs in-memory by neighborhood_name.lower(), submarket_cluster.lower(), zipcode, type_of_space.
2. Select O_Search queryset by ids/filter/since; fetch only needed fields via .only().
3. For each O_Search:
   • Normalize keys: neighborhood_key = submarket_name or submarket_cluster, zipcode_i = int(zipcode) if possible.
   • Find exact LP matches (neighborhood & type), then zipcode-filtered subset if present.
   • Decide flags per rules; compute the tuple of all status ints.
   • If --update-m2m: compute desired pages/pages2/pages3/pages4 sets and diff against current relations.
   • Only stage updates where any value would change (idempotent).
4. Bulk write updates within transactions; commit per chunk.

Data updates

• Fields potentially written on O_Search:
  • status_direct, status_sublet, status_coworking, status_nearby
  • cluster_status_*, industry_status_*, zipcode_status_*, manhattan_status
  • updated_at auto-updated
• If --update-m2m:
  • pages: exact neighborhood/type matches
  • pages2: cluster matches
  • pages3: industry (space_use) matches
  • pages4: “all Manhattan” or global pages

Command skeleton

# app/management/commands/recompute_statuses.py
from django.core.management.base import BaseCommand
from django.db import transaction
from django.utils import timezone
from django.db.models import Q
from app.models import O_Search, Landing_Page

class Command(BaseCommand):
    help = "Recompute O_Search status flags based on Landing_Page constraints and relations."

    def add_arguments(self, parser):
        parser.add_argument("--ids")
        parser.add_argument("--filter")
        parser.add_argument("--since")
        parser.add_argument("--chunk-size", type=int, default=500)
        parser.add_argument("--dry-run", action="store_true")
        parser.add_argument("--update-m2m", action="store_true")
        parser.add_argument("--language")

    def handle(self, *args, **opts):
        lps = self._load_active_pages(opts.get("language"))
        qs  = self._build_queryset(opts)

        updated = 0
        for batch in self._iter(qs, opts["chunk_size"]):
            changes = [self._decide(o, lps, opts) for o in batch]
            to_write = [c for c in changes if c is not None]

            if not opts["dry_run"] and to_write:
                with transaction.atomic():
                    for obj, vals, m2m in to_write:
                        for f, v in vals.items():
                            setattr(obj, f, v)
                        obj.save(update_fields=list(vals.keys()))
                        if opts["update_m2m"]:
                            self._apply_m2m(obj, m2m)
            updated += len(to_write)

        self.stdout.write(f"updated={updated} dry_run={opts['dry_run']}")

    def _load_active_pages(self, language):
        lp = Landing_Page.objects.filter(state=True)
        if language:
            lp = lp.filter(language=language)
        # Pre-index by keys for O(1) lookups
        idx = {
            "by_neighborhood": {},
            "by_cluster": {},
            "by_zip": {},
            "list": list(lp.only("id","neighborhood_name","submarket_cluster","zipcode","type_of_space","nearby_landing_page_ids"))
        }
        for p in idx["list"]:
            if p.neighborhood_name:
                idx["by_neighborhood"].setdefault(p.neighborhood_name.lower(), []).append(p)
            if p.submarket_cluster:
                idx["by_cluster"].setdefault(p.submarket_cluster.lower(), []).append(p)
            if p.zipcode:
                idx["by_zip"].setdefault(int(p.zipcode), []).append(p)
        return idx

    def _build_queryset(self, opts):
        qs = O_Search.objects.all().only(
            "id","submarket_name","submarket_cluster","zipcode","space_type","space_use",
            "status_direct","status_sublet","status_coworking","status_nearby",
            "cluster_status_direct","cluster_status_sublet","cluster_status_coworking","cluster_status_nearby",
            "industry_status_direct","industry_status_sublet","industry_status_coworking","industry_status_nearby",
            "zipcode_status_direct","zipcode_status_sublet","zipcode_status_coworking","zipcode_status_nearby",
            "manhattan_status"
        ).order_by("id")
        if opts.get("ids"):
            qs = qs.filter(id__in=[int(x) for x in opts["ids"].split(",")])
        if opts.get("filter"):
            # Simple k=v,k=v (safe subset only)
            for expr in opts["filter"].split(","):
                k, v = expr.split("=")
                qs = qs.filter(**{k: v})
        if opts.get("since"):
            qs = qs.filter(updated_at__gte=opts["since"])
        return qs

    def _iter(self, qs, size):
        start = 0
        while True:
            batch = list(qs[start:start+size])
            if not batch: break
            yield batch
            start += size

    def _compat(self, lp, o):
        # Space type compatibility (extend as needed)
        if lp.type_of_space and o.space_type:
            return lp.type_of_space.strip().lower() == o.space_type.strip().lower()
        return True

    def _decide(self, o, idx, opts):
        key = (o.submarket_name or o.submarket_cluster or "").strip().lower()
        try:
            z = int(o.zipcode) if o.zipcode and str(o.zipcode).isdigit() else None
        except Exception:
            z = None

        exact = (idx["by_neighborhood"].get(key, []) or idx["by_cluster"].get(key, []))
        if z is not None:
            exact = [p for p in exact if (p.zipcode is None or int(p.zipcode) == z)]

        direct = any(self._compat(p, o) for p in exact)
        sublet = (o.space_type or "").lower().find("sublet") >= 0 and any(self._compat(p, o) for p in exact)
        cowork = (o.space_type or "").lower() == "coworking" and any(self._compat(p, o) for p in exact)

        nearby = False
        m2m = {"pages": set(), "pages2": set(), "pages3": set(), "pages4": set()}
        if not direct:
            # Expand via nearby_landing_page_ids of first compatible cluster page
            for p in exact or idx["by_cluster"].get(key, []):
                if not self._compat(p, o): continue
                if p.nearby_landing_page_ids:
                    ids = [int(x) for x in str(p.nearby_landing_page_ids).split(",") if x.strip().isdigit()]
                    m2m["pages"].add(p.id)
                    m2m["pages2"].update(ids)
                    nearby = True
                    break

        vals = {
            "status_direct": 1 if direct else 0,
            "status_sublet": 1 if sublet else 0,
            "status_coworking": 1 if cowork else 0,
            "status_nearby": 1 if nearby else 0,
            # Rollups
            "cluster_status_direct": 1 if direct else 0,
            "cluster_status_sublet": 1 if sublet else 0,
            "cluster_status_coworking": 1 if cowork else 0,
            "cluster_status_nearby": 1 if nearby else 0,
            "industry_status_direct": 1 if direct and (o.space_use or "").strip() else 0,
            "industry_status_sublet": 1 if sublet and (o.space_use or "").strip() else 0,
            "industry_status_coworking": 1 if cowork and (o.space_use or "").strip() else 0,
            "industry_status_nearby": 1 if nearby and (o.space_use or "").strip() else 0,
            "zipcode_status_direct": 1 if direct and z is not None else 0,
            "zipcode_status_sublet": 1 if sublet and z is not None else 0,
            "zipcode_status_coworking": 1 if cowork and z is not None else 0,
            "zipcode_status_nearby": 1 if nearby and z is not None else 0,
            "manhattan_status": 1 if ("manhattan" in (o.submarket_cluster or "").lower()) else 0,
        }

        changed = any(getattr(o, f) != v for f, v in vals.items())
        if not changed and not opts.get("update_m2m"):
            return None
        return (o, vals, m2m)

    def _apply_m2m(self, o, m2m):
        if m2m["pages"]:
            o.pages.set(list(m2m["pages"]))  # direct/neighborhood pages
        if m2m["pages2"]:
            o.pages2.set(list(m2m["pages2"]))  # cluster/nearby
        # pages3/pages4 can be filled by industry/global rules as needed

Indices & performance

• O_Search(submarket_name), O_Search(submarket_cluster), O_Search(zipcode), O_Search(space_type)
• Landing_Page(state), Landing_Page(neighborhood_name), Landing_Page(submarket_cluster), Landing_Page(zipcode), Landing_Page(type_of_space)
• Chunked processing, .only() field selection, pre-index Landing Pages in memory.

Testing

• Unit: matching logic (neighborhood vs cluster, zipcode filter, type compatibility).
• Unit: sublet/coworking detection, nearby expansion from nearby_landing_page_ids.
• Integration: round-trip with M2M updates; idempotent re-run → zero changes.

Acceptance criteria

• Dry run shows correct diffs; real run updates only changed rows.
• Flags set consistently based on Landing Page availability and compatibility.
• M2M links reflect direct/nearby placement when --update-m2m is used.

Note: adjust compatibility mapping (e.g., normalize type_of_space vocab) and any Manhattan/industry heuristics to your production rules.