Scope monitoring PF objects to Telegram-linked users

app/bot.py

@@ -31,12 +31,20 @@ logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name
 logger = logging.getLogger(__name__)
 
 
-async def cmd_start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
+def _portal_user_code(context: ContextTypes.DEFAULT_TYPE) -> str | None:
+    if not context.args:
+        return None
+    code = context.args[0].strip()
+    return code or None
+
+
+async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
     if not update.effective_user or not update.effective_chat:
         return
     user = update.effective_user
     chat_id = str(update.effective_chat.id)
     username = user.username
+    portal_user_id = _portal_user_code(context)
 
     db = SessionLocal()
     try:
@@ -44,6 +52,21 @@ async def cmd_start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
             db.query(Employee).filter(Employee.tg_chat_id == chat_id).first()
         )
         if existing:
+            if portal_user_id and existing.portal_user_id and existing.portal_user_id != portal_user_id:
+                await update.message.reply_text(
+                    "Этот Telegram уже подключен к другому пользователю Portal.",
+                )
+                return
+            if portal_user_id and not existing.portal_user_id:
+                clash = db.query(Employee).filter(Employee.portal_user_id == portal_user_id).first()
+                if clash and clash.id != existing.id:
+                    await update.message.reply_text(
+                        "Этот пользователь Portal уже подключен к другому Telegram.",
+                    )
+                    return
+                existing.portal_user_id = portal_user_id
+                existing.tg_username = username
+                db.commit()
             await update.message.reply_text(
                 f"✅ Вы уже подключены как <b>{existing.name}</b>.\n"
                 f"chat_id: <code>{chat_id}</code>",
@@ -51,6 +74,34 @@ async def cmd_start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
             )
             return
 
+        if portal_user_id:
+            employee = db.query(Employee).filter(Employee.portal_user_id == portal_user_id).first()
+            name = (user.full_name or username or f"user_{chat_id}").strip()
+            if employee:
+                if employee.tg_chat_id and employee.tg_chat_id != chat_id:
+                    await update.message.reply_text(
+                        "Этот пользователь Portal уже подключен к другому Telegram.",
+                    )
+                    return
+                employee.name = employee.name or name
+                employee.tg_chat_id = chat_id
+                employee.tg_username = username
+            else:
+                employee = Employee(
+                    name=name,
+                    portal_user_id=portal_user_id,
+                    tg_chat_id=chat_id,
+                    tg_username=username,
+                )
+                db.add(employee)
+            db.commit()
+            await update.message.reply_text(
+                f"✅ Привет, <b>{name}</b>! Telegram подключен к вашему аккаунту Portal.\n"
+                f"Теперь можно добавлять объекты мониторинга в Portal.",
+                parse_mode="HTML",
+            )
+            return
+
         # Try to find by username (admin pre-created employee w/o chat_id)
         if username:
             placeholder = (
@@ -68,15 +119,10 @@ async def cmd_start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
                 )
                 return
 
-        # Create a new employee record from this user
-        name = (user.full_name or username or f"user_{chat_id}").strip()
-        e = Employee(name=name, tg_chat_id=chat_id, tg_username=username)
-        db.add(e)
-        db.commit()
         await update.message.reply_text(
-            f"👋 Привет, <b>{name}</b>! Вы зарегистрированы как сотрудник.\n"
+            "Откройте Portal → Мониторинг PF и нажмите подключение Telegram.\n"
-            f"Откройте веб-интерфейс и создайте проекты, чтобы получать уведомления.\n"
+            "Бот должен получить команду вида:\n"
-            f"chat_id: <code>{chat_id}</code>",
+            "<code>/start ваш_код_из_Portal</code>",
             parse_mode="HTML",
         )
     finally:

app/config.py

@@ -34,6 +34,7 @@ class Settings(BaseSettings):
     scrape_interval_hours: int = 4
     database_url: str = f"sqlite:///{DATA_DIR / 'monitor.db'}"
     admin_chat_id: str = ""
+    tg_bot_username: str = ""
 
     def model_post_init(self, __context) -> None:
         self.database_url = _resolve_sqlite_url(self.database_url)

app/db.py

@@ -1,4 +1,4 @@
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect, text
 from sqlalchemy.orm import DeclarativeBase, sessionmaker
 
 from app.config import settings
@@ -29,3 +29,17 @@ def init_db():
     from app import models  # noqa: F401 — registers models on Base
 
     Base.metadata.create_all(bind=engine)
+    _migrate_employees_portal_user_id()
+
+
+def _migrate_employees_portal_user_id() -> None:
+    inspector = inspect(engine)
+    if "employees" not in inspector.get_table_names():
+        return
+    columns = {col["name"] for col in inspector.get_columns("employees")}
+    with engine.begin() as conn:
+        if "portal_user_id" not in columns:
+            conn.execute(text("ALTER TABLE employees ADD COLUMN portal_user_id VARCHAR(100)"))
+        conn.execute(
+            text("CREATE UNIQUE INDEX IF NOT EXISTS ix_employees_portal_user_id ON employees (portal_user_id)")
+        )

app/models.py

@@ -27,6 +27,7 @@ class Employee(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     name: Mapped[str] = mapped_column(String(200))
+    portal_user_id: Mapped[str | None] = mapped_column(String(100), unique=True, index=True, nullable=True)
     tg_chat_id: Mapped[str | None] = mapped_column(String(64), unique=True, nullable=True)
     tg_username: Mapped[str | None] = mapped_column(String(200), nullable=True)
     created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

app/web.py

@@ -30,6 +30,7 @@ app = FastAPI(title="Monitoring PF")
 
 class EmployeeCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=200)
+    portal_user_id: str | None = Field(None, max_length=100)
     tg_username: str | None = Field(None, max_length=200)
     tg_chat_id: str | None = Field(None, max_length=64)
 
@@ -43,7 +44,7 @@ class EmployeeUpdate(BaseModel):
 class ProjectCreate(BaseModel):
     title: str = Field(..., min_length=1, max_length=300)
     deal_type: DealType
-    owner_id: int
+    owner_id: int | None = None
     our_price: float | None = None
     notes: str | None = None
     dld_permit: str | None = Field(None, max_length=100)
@@ -103,6 +104,49 @@ def _clean(value: str | None) -> str | None:
     return value or None
 
 
+def _portal_user_id(request: Request) -> str | None:
+    return _clean(request.headers.get("x-user-id"))
+
+
+def _telegram_start_command(portal_user_id: str | None) -> str | None:
+    return f"/start {portal_user_id}" if portal_user_id else None
+
+
+def _telegram_start_link(portal_user_id: str | None) -> str | None:
+    username = settings.tg_bot_username.strip().lstrip("@")
+    if not username or not portal_user_id:
+        return None
+    return f"https://t.me/{username}?start={portal_user_id}"
+
+
+def _current_employee(request: Request, db: Session, *, required: bool = True) -> Employee | None:
+    portal_user_id = _portal_user_id(request)
+    if not portal_user_id:
+        if required:
+            raise HTTPException(status_code=401, detail="portal user is not available")
+        return None
+    employee = db.query(Employee).filter(Employee.portal_user_id == portal_user_id).first()
+    if (not employee or not employee.tg_chat_id) and required:
+        raise HTTPException(status_code=403, detail="Сначала авторизуйтесь в Telegram-боте Monitoring PF")
+    return employee
+
+
+def _owned_project(request: Request, db: Session, project_id: int, *, with_detail: bool = False) -> Project:
+    employee = _current_employee(request, db)
+    query = db.query(Project).filter(Project.id == project_id, Project.owner_id == employee.id)
+    if with_detail:
+        query = query.options(
+            joinedload(Project.owner),
+            joinedload(Project.listings).joinedload(CompetitorListing.price_history),
+        )
+    else:
+        query = query.options(joinedload(Project.owner), joinedload(Project.listings))
+    project = query.first()
+    if not project:
+        raise HTTPException(404, "project not found")
+    return project
+
+
 def _dt(value: datetime | None) -> str | None:
     return value.isoformat() if value else None
 
@@ -111,6 +155,7 @@ def _employee_out(employee: Employee) -> dict[str, Any]:
     return {
         "id": employee.id,
         "name": employee.name,
+        "portal_user_id": employee.portal_user_id,
         "tg_chat_id": employee.tg_chat_id,
         "tg_username": employee.tg_username,
         "projects_total": len(employee.projects or []),
@@ -197,15 +242,41 @@ def _suggestion_out(item: Any) -> dict[str, Any]:
 
 
 @app.get("/api/v1/access/me")
-def access_me(request: Request) -> dict[str, Any]:
+def access_me(request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
-    return {"is_admin": _is_admin(request)}
+    portal_user_id = _portal_user_id(request)
+    employee = _current_employee(request, db, required=False)
+    return {
+        "is_admin": _is_admin(request),
+        "portal_user_id": portal_user_id,
+        "telegram_linked": bool(employee and employee.tg_chat_id),
+        "employee": _employee_out(employee) if employee else None,
+        "telegram_bot_username": settings.tg_bot_username.strip().lstrip("@") or None,
+        "telegram_start_command": _telegram_start_command(portal_user_id),
+        "telegram_start_link": _telegram_start_link(portal_user_id),
+    }
 
 
 @app.get("/api/v1/summary")
-def summary(db: Session = Depends(get_db)) -> dict[str, Any]:
+def summary(request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
-    projects = db.query(Project).options(joinedload(Project.listings)).all()
+    employee = _current_employee(request, db, required=False)
-    employees = db.query(Employee).all()
+    if not employee or not employee.tg_chat_id:
-    listings = db.query(CompetitorListing).all()
+        projects = []
+        employees = []
+        listings = []
+    else:
+        projects = (
+            db.query(Project)
+            .options(joinedload(Project.listings))
+            .filter(Project.owner_id == employee.id)
+            .all()
+        )
+        employees = [employee]
+        listings = (
+            db.query(CompetitorListing)
+            .join(Project)
+            .filter(Project.owner_id == employee.id)
+            .all()
+        )
     active = [l for l in listings if l.status == ListingStatus.ACTIVE]
     return {
         "projects_total": len(projects),
@@ -219,20 +290,30 @@ def summary(db: Session = Depends(get_db)) -> dict[str, Any]:
 
 
 @app.get("/api/v1/employees")
-def employees_list(db: Session = Depends(get_db)) -> list[dict[str, Any]]:
+def employees_list(request: Request, db: Session = Depends(get_db)) -> list[dict[str, Any]]:
-    employees = (
+    if _is_admin(request):
-        db.query(Employee)
+        employees = (
-        .options(joinedload(Employee.projects))
+            db.query(Employee)
-        .order_by(Employee.name)
+            .options(joinedload(Employee.projects))
-        .all()
+            .order_by(Employee.name)
-    )
+            .all()
+        )
+    else:
+        employee = _current_employee(request, db, required=False)
+        employees = [employee] if employee and employee.tg_chat_id else []
     return [_employee_out(employee) for employee in employees]
 
 
 @app.post("/api/v1/employees", status_code=201)
-def employee_create(payload: EmployeeCreate, db: Session = Depends(get_db)) -> dict[str, Any]:
+def employee_create(
+    payload: EmployeeCreate,
+    request: Request,
+    db: Session = Depends(get_db),
+) -> dict[str, Any]:
+    _require_admin(request)
     employee = Employee(
         name=payload.name.strip(),
+        portal_user_id=_clean(payload.portal_user_id),
         tg_username=_clean(payload.tg_username).lstrip("@") if _clean(payload.tg_username) else None,
         tg_chat_id=_clean(payload.tg_chat_id),
     )
@@ -286,10 +367,12 @@ def employee_delete(employee_id: int, request: Request, db: Session = Depends(ge
 
 
 @app.get("/api/v1/projects")
-def projects_list(db: Session = Depends(get_db)) -> list[dict[str, Any]]:
+def projects_list(request: Request, db: Session = Depends(get_db)) -> list[dict[str, Any]]:
+    employee = _current_employee(request, db)
     projects = (
         db.query(Project)
         .options(joinedload(Project.owner), joinedload(Project.listings))
+        .filter(Project.owner_id == employee.id)
         .order_by(Project.created_at.desc())
         .all()
     )
@@ -297,10 +380,8 @@ def projects_list(db: Session = Depends(get_db)) -> list[dict[str, Any]]:
 
 
 @app.post("/api/v1/projects", status_code=201)
-def project_create(payload: ProjectCreate, db: Session = Depends(get_db)) -> dict[str, Any]:
+def project_create(payload: ProjectCreate, request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
-    owner = db.get(Employee, payload.owner_id)
+    owner = _current_employee(request, db)
-    if not owner:
-        raise HTTPException(404, "employee not found")
     project = Project(
         title=payload.title.strip(),
         deal_type=payload.deal_type,
@@ -320,67 +401,56 @@ def project_create(payload: ProjectCreate, db: Session = Depends(get_db)) -> dic
 
 
 @app.get("/api/v1/projects/{project_id}")
-def project_detail(project_id: int, db: Session = Depends(get_db)) -> dict[str, Any]:
+def project_detail(project_id: int, request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
-    project = (
+    project = _owned_project(request, db, project_id, with_detail=True)
-        db.query(Project)
-        .options(
-            joinedload(Project.owner),
-            joinedload(Project.listings).joinedload(CompetitorListing.price_history),
-        )
-        .filter(Project.id == project_id)
-        .first()
-    )
-    if not project:
-        raise HTTPException(404, "project not found")
     return _project_out(project, detail=True)
 
 
 @app.patch("/api/v1/projects/{project_id}")
-def project_update(project_id: int, payload: ProjectUpdate, db: Session = Depends(get_db)) -> dict[str, Any]:
+def project_update(
-    project = db.get(Project, project_id)
+    project_id: int,
-    if not project:
+    payload: ProjectUpdate,
-        raise HTTPException(404, "project not found")
+    request: Request,
+    db: Session = Depends(get_db),
+) -> dict[str, Any]:
+    project = _owned_project(request, db, project_id)
     data = payload.model_dump(exclude_unset=True)
-    if "owner_id" in data and data["owner_id"] is not None:
+    data.pop("owner_id", None)
-        owner = db.get(Employee, data["owner_id"])
-        if not owner:
-            raise HTTPException(404, "employee not found")
-        project.owner_id = owner.id
     for field in ("title", "deal_type", "our_price", "notes", "dld_permit", "building", "bedrooms", "size_sqft", "our_url"):
-        if field not in data or field == "owner_id":
+        if field not in data:
             continue
         value = data[field]
         if isinstance(value, str):
             value = _clean(value)
         setattr(project, field, value)
     db.commit()
-    return project_detail(project_id, db)
+    db.refresh(project)
+    return _project_out(_owned_project(request, db, project_id, with_detail=True), detail=True)
 
 
 @app.delete("/api/v1/projects/{project_id}", status_code=204)
 def project_delete(project_id: int, request: Request, db: Session = Depends(get_db)) -> None:
-    _require_admin(request)
+    project = _owned_project(request, db, project_id)
-    project = db.get(Project, project_id)
-    if not project:
-        raise HTTPException(404, "project not found")
     db.delete(project)
     db.commit()
 
 
 @app.post("/api/v1/projects/{project_id}/check")
-def project_check_now(project_id: int, db: Session = Depends(get_db)) -> dict[str, int]:
+def project_check_now(project_id: int, request: Request, db: Session = Depends(get_db)) -> dict[str, int]:
-    if not db.get(Project, project_id):
+    _owned_project(request, db, project_id)
-        raise HTTPException(404, "project not found")
     db.close()
     changes = run_check_for_project(project_id)
     return {"changes": changes}
 
 
 @app.post("/api/v1/projects/{project_id}/listings", status_code=201)
-def listing_create(project_id: int, payload: ListingCreate, db: Session = Depends(get_db)) -> dict[str, Any]:
+def listing_create(
-    project = db.get(Project, project_id)
+    project_id: int,
-    if not project:
+    payload: ListingCreate,
-        raise HTTPException(404, "project not found")
+    request: Request,
+    db: Session = Depends(get_db),
+) -> dict[str, Any]:
+    project = _owned_project(request, db, project_id)
     listing, err = add_competitor_url(db, project, payload.url)
     if err:
         raise HTTPException(400, err)
@@ -388,17 +458,25 @@ def listing_create(project_id: int, payload: ListingCreate, db: Session = Depend
 
 
 @app.post("/api/v1/projects/{project_id}/listings/bulk")
-def listings_bulk(project_id: int, payload: ListingsBulkCreate, db: Session = Depends(get_db)) -> dict[str, Any]:
+def listings_bulk(
-    project = db.get(Project, project_id)
+    project_id: int,
-    if not project:
+    payload: ListingsBulkCreate,
-        raise HTTPException(404, "project not found")
+    request: Request,
+    db: Session = Depends(get_db),
+) -> dict[str, Any]:
+    project = _owned_project(request, db, project_id)
     return add_competitor_urls(db, project, payload.urls)
 
 
 @app.delete("/api/v1/listings/{listing_id}", status_code=204)
 def listing_delete(listing_id: int, request: Request, db: Session = Depends(get_db)) -> None:
-    _require_admin(request)
+    employee = _current_employee(request, db)
-    listing = db.get(CompetitorListing, listing_id)
+    listing = (
+        db.query(CompetitorListing)
+        .join(Project)
+        .filter(CompetitorListing.id == listing_id, Project.owner_id == employee.id)
+        .first()
+    )
     if not listing:
         raise HTTPException(404, "listing not found")
     db.delete(listing)
@@ -406,15 +484,8 @@ def listing_delete(listing_id: int, request: Request, db: Session = Depends(get_
 
 
 @app.get("/api/v1/projects/{project_id}/suggest")
-def project_suggest(project_id: int, db: Session = Depends(get_db)) -> dict[str, Any]:
+def project_suggest(project_id: int, request: Request, db: Session = Depends(get_db)) -> dict[str, Any]:
-    project = (
+    project = _owned_project(request, db, project_id)
-        db.query(Project)
-        .options(joinedload(Project.listings))
-        .filter(Project.id == project_id)
-        .first()
-    )
-    if not project:
-        raise HTTPException(404, "project not found")
     permit = resolve_our_permit(project)
     suggestions = suggest_similar(project, our_permit=permit)
     return {

k8s/configmap.yaml

@@ -9,3 +9,4 @@ data:
   PUBLIC_BASE_PATH: "/api/monitoring-pf"
   DATABASE_URL: "sqlite:////data/monitor.db"
   SCRAPE_INTERVAL_HOURS: "4"
+  TG_BOT_USERNAME: ""