סיימתי את זה

Signed-off-by: Dmitri Shimanski <dshiafeed@gmail.com>

db_repository/BaseRepositories.py

@@ -0,0 +1,131 @@
+import sqlite3
+from typing import Type, Optional, List
+from pydantic import BaseModel
+from threading import Lock
+
+
+class BaseRepository:
+    def __init__(self, mutex: Lock, db_path: str = "app.db"):
+        self.mutex = mutex
+        self.db_path = db_path
+        self._init_db()
+
+    @property
+    def _connection(self) -> sqlite3.Connection:
+        """Создает новое соединение для каждого запроса"""
+        return sqlite3.connect(self.db_path)
+
+    def _init_db(self):
+        """Инициализация таблиц в базе данных"""
+        with self.mutex:
+            with self._connection as conn:
+                cursor = conn.cursor()
+                statements = self.create_table_sql.split(';')
+                for stmt in statements:
+                    stmt = stmt.strip()
+                    if stmt:
+                        cursor.execute(stmt)
+                conn.commit()
+
+    @property
+    def create_table_sql(self) -> str:
+        """Должен быть переопределен в дочерних классах"""
+        raise NotImplementedError
+
+    def _execute(
+            self,
+            sql: str,
+            params: tuple = (),
+            fetch: bool = False
+    ) -> Optional[List[dict]]:
+        """Общий метод для выполнения запросов с возвратом словарей"""
+        with self._connection as conn:
+            cursor = conn.cursor()
+            cursor.execute(sql, params)
+
+            if fetch:
+                columns = [col[0] for col in cursor.description]
+                results = [dict(zip(columns, row)) for row in cursor.fetchall()]
+                return results
+
+            conn.commit()
+            return None
+
+
+class BaseModelSchema(BaseModel):
+    id: Optional[int] = None
+
+
+class BaseCRUDRepository(BaseRepository):
+    table_name: str = ""
+    schema: Type[BaseModelSchema] = BaseModelSchema
+
+    @property
+    def create_table_sql(self) -> str:
+        return f"""
+        CREATE TABLE IF NOT EXISTS {self.table_name} (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            {self._get_columns_definition()}
+        );
+        """
+
+    def _get_columns_definition(self) -> str:
+        """Генерирует SQL-определение колонок на основе модели Pydantic"""
+        fields = self.schema.model_fields
+        columns = []
+        for name, field in fields.items():
+            if name == "id":
+                continue
+            sql_type = "TEXT" if field.annotation == str else "INTEGER"
+            nullable = "NOT NULL" if not field.is_required else ""
+            columns.append(f"{name} {sql_type} {nullable}")
+        return ", ".join(columns)
+
+    def create(self, item: BaseModelSchema) -> int:
+        fields = item.dict(exclude={"id"})
+        columns = ", ".join(fields.keys())
+        placeholders = ", ".join(["?"] * len(fields))
+
+        sql = f"""
+        INSERT INTO {self.table_name} ({columns})
+        VALUES ({placeholders})
+        """
+        self._execute(sql, tuple(fields.values()))
+        return self._get_last_insert_id()
+
+    def get(self, item_id: int) -> Optional[BaseModelSchema]:
+        sql = f"SELECT * FROM {self.table_name} WHERE id = ?"
+        result = self._execute(sql, (item_id,), fetch=True)
+        return self.schema(**dict(result[0])) if result else None
+
+    def get_by_name(self, item_id: str) -> Optional[BaseModelSchema]:
+        sql = f"SELECT * FROM {self.table_name} WHERE name = ?"
+        result = self._execute(sql, (item_id,), fetch=True)
+        return self.schema(**dict(result[0])) if result else None
+
+    def get_all(self) -> List[BaseModelSchema]:
+        sql = f"SELECT * FROM {self.table_name}"
+        results = self._execute(sql, fetch=True)
+        return [self.schema(**dict(row)) for row in results]
+
+    def update(self, item_id: int, item: BaseModelSchema) -> bool:
+        fields = item.dict(exclude={"id"})
+        set_clause = ", ".join([f"{key} = ?" for key in fields.keys()])
+
+        sql = f"""
+        UPDATE {self.table_name}
+        SET {set_clause}
+        WHERE id = ?
+        """
+        params = (*fields.values(), item_id)
+        self._execute(sql, params)
+        return True
+
+    def delete(self, item_id: int) -> bool:
+        sql = f"DELETE FROM {self.table_name} WHERE id = ?"
+        self._execute(sql, (item_id,))
+        return True
+
+    def _get_last_insert_id(self) -> int:
+        result = self._execute("SELECT last_insert_rowid()", fetch=True)
+        return result[0]["last_insert_rowid()"] if result else 0

db_repository/Volunteer.py

@@ -0,0 +1,14 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from db_repository.BaseRepositories import BaseCRUDRepository
+from models.Volunteer import Volunteer
+
+
+class VolunteerCreate(BaseModel, Volunteer):
+    id: int
+
+
+class VolunteerSchema(VolunteerCreate):
+    id: int = None

db_repository/apartment_repository.py

@@ -0,0 +1,67 @@
+import array
+from typing import Optional, List
+
+from db_repository.BaseRepositories import BaseCRUDRepository
+from db_repository.schemas.apartment import ApartmentSchema
+import json
+
+
+class ApartmentRepository(BaseCRUDRepository):
+    table_name = "apartments"
+    schema = ApartmentSchema
+
+    def _get_columns_definition(self) -> str:
+        base_columns = super()._get_columns_definition()
+        # Добавляем кастомные поля, которые не обрабатываются автоматически
+        return (
+            "owner TEXT NOT NULL, "
+            "location TEXT NOT NULL, "
+            "rooms INTEGER NOT NULL, "
+            "has_mamad TEXT NOT NULL, "
+            "price TEXT NOT NULL, "
+            "accepted_regions TEXT NOT NULL, "
+            "is_available INTEGER NOT NULL"
+        )
+
+    def create(self, item: ApartmentSchema) -> int:
+        item_dict = item.model_dump()
+        return super().create(self.schema(**item_dict))
+
+    def get(self, item_id: int) -> Optional[ApartmentSchema]:
+        result = super().get(item_id)
+        if result:
+            result_dict = result.dict()
+            result_dict["accepted_regions"] = result_dict["accepted_regions"]
+            return ApartmentSchema(**result_dict)
+        return None
+
+    def search_available(
+            self,
+            region: str,
+            min_rooms: int = 0,
+    ) -> List[ApartmentSchema]:
+        query = f"""
+        SELECT * FROM {self.table_name}
+        WHERE is_available = 1 
+          AND rooms >= ? 
+        """
+        params = (min_rooms,)
+
+        results = self._execute(query, params, fetch=True)
+        return [
+            self._parse_result(row)
+            for row in results
+            if (region in row["accepted_regions"] or "all" in row["accepted_regions"])
+        ]
+
+    def _parse_result(self, row: tuple) -> ApartmentSchema:
+        return ApartmentSchema(
+            id=row["id"],
+            owner=row["owner"],
+            location=row["location"],
+            rooms=row["rooms"],
+            has_mamad=row["has_mamad"],
+            price=row["price"],
+            accepted_regions=row["accepted_regions"],
+            is_available=bool(row["is_available"]),
+        )

db_repository/needy_repository.py

@@ -0,0 +1,22 @@
+from typing import List
+
+from db_repository.BaseRepositories import BaseCRUDRepository
+from db_repository.schemas.needy import NeedySchema
+
+
+class NeedyRepository(BaseCRUDRepository):
+    table_name = "needies"
+    schema = NeedySchema
+
+    def _get_columns_definition(self) -> str:
+        return (
+            "name TEXT NOT NULL, "
+            "contact TEXT NOT NULL, "
+            "region TEXT NOT NULL, "
+            "how_much_peoples INTEGER NOT NULL"
+        )
+
+    def get_by_region(self, region: str) -> List[NeedySchema]:
+        query = f"SELECT * FROM {self.table_name} WHERE region = ?"
+        results = self._execute(query, (region,), fetch=True)
+        return [self.schema(**dict(row)) for row in results]

db_repository/schemas/apartment.py

@@ -0,0 +1,12 @@
+from .base import BaseSchema
+from typing import List
+
+
+class ApartmentSchema(BaseSchema):
+    owner: str
+    location: str
+    rooms: int
+    has_mamad: str
+    price: str
+    accepted_regions: str
+    is_available: bool = True

db_repository/schemas/base.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+from typing import List, Optional
+
+
+class BaseSchema(BaseModel):
+    id: Optional[int] = None

db_repository/schemas/needy.py

@@ -0,0 +1,6 @@
+from .user import UserSchema
+
+
+class NeedySchema(UserSchema):
+    region: str
+    how_much_peoples: int

db_repository/schemas/user.py

@@ -0,0 +1,6 @@
+from .base import BaseSchema
+
+
+class UserSchema(BaseSchema):
+    name: str
+    contact: str

db_repository/schemas/volunteer.py

@@ -0,0 +1,5 @@
+from .user import UserSchema
+
+
+class VolunteerSchema(UserSchema):
+    pass

db_repository/volunteer_repository.py

@@ -0,0 +1,45 @@
+from db_repository.BaseRepositories import BaseCRUDRepository
+from db_repository.schemas.volunteer import VolunteerSchema
+from typing import List
+
+
+class VolunteerRepository(BaseCRUDRepository):
+    table_name = "volunteers"
+    schema = VolunteerSchema
+
+    def _get_columns_definition(self) -> str:
+        return (
+            "name TEXT NOT NULL, "
+            "contact TEXT NOT NULL"
+        )
+
+    @property
+    def create_table_sql(self) -> str:
+        return f"""
+            CREATE TABLE IF NOT EXISTS {self.table_name} (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                {self._get_columns_definition()}
+            );
+
+            CREATE TABLE IF NOT EXISTS volunteer_apartments (
+                volunteer_id INTEGER NOT NULL,
+                apartment_id INTEGER NOT NULL,
+                FOREIGN KEY(volunteer_id) REFERENCES volunteers(id),
+                FOREIGN KEY(apartment_id) REFERENCES apartments(id)
+            );
+            """
+
+    def add_apartment(self, volunteer_id: int, apartment_id: int):
+        query = """
+        INSERT INTO volunteer_apartments (volunteer_id, apartment_id)
+        VALUES (?, ?)
+        """
+        self._execute(query, (volunteer_id, apartment_id))
+
+    def get_apartments(self, volunteer_id: int) -> List[int]:
+        query = """
+        SELECT apartment_id FROM volunteer_apartments
+        WHERE volunteer_id = ?
+        """
+        results = self._execute(query, (volunteer_id,), fetch=True)
+        return [row[0] for row in results] if results else []

main.py

@@ -0,0 +1,67 @@
+from models.Apps import HousingApp
+
+
+def main():
+    app = HousingApp()
+
+    while True:
+        print("\n1. Register as Volunteer")
+        print("2. Register as Evacuee")
+        print("3. Add Apartment")
+        print("4. Search Apartments")
+        print("5. Exit")
+
+        choice = input("Choose option: ")
+
+        if choice == '1':
+            name = input("Full name: ")
+            contact = input("Contact info: ")
+            app.register_volunteer(name, contact)
+            print("Volunteer registered successfully!")
+
+        elif choice == '2':
+            name = input("Full name: ")
+            contact = input("Contact info: ")
+            region = input("Region (north/south): ").lower()
+            how_much_peoples = int(input("How many people in family?: "))
+            app.register_evacuee(name, contact, region, how_much_peoples)
+            print("Evacuee registered successfully!")
+
+        elif choice == '3':
+            name = input("Volunteer name: ")
+            volunteer = next((v for v in app.volunteers if v.name == name), None)
+            if volunteer:
+                location = input("Apartment location: ")
+                rooms = int(input("Number of rooms: "))
+                has_mamad = input("Has safe room? (yes/no): ").lower() == 'yes'
+                price = input("Payment terms: ")
+                regions = input("Accepts residents from (north/south/all): ")
+                app.add_apartment(volunteer.name, location, rooms, has_mamad, price, regions)
+                print("Apartment added to system!")
+
+        elif choice == '4':
+            name = input("Evacuee name: ")
+            evacuee = next((e for e in app.needys if e.name == name), None)
+            if evacuee:
+                apartments = app.search_apartments(evacuee.name)
+                if apartments:
+                    print("\nAvailable apartments:")
+                    for i, apt in enumerate(apartments, 1):
+                        print(f"{i}. {apt}")
+                    selection = int(input("Select apartment (number): ")) - 1
+                    if app.book_apartment(apartments[selection].id, evacuee.name):
+                        print("Apartment booked successfully!")
+                    else:
+                        print("Apartment no longer available")
+                else:
+                    print("No matching apartments found")
+
+        elif choice == '5':
+            print("Thank you for using the Housing Match App!")
+            break
+        else:
+            print("Write, please only numbers, provided in start message")
+
+
+if __name__ == "__main__":
+    main()

models/Apartment.py

@@ -1,15 +1,14 @@
 class Apartment:
-    def __init__(self, 
+    def __init__(self, owner, location, rooms, has_mamad, price, accepted_regions):
-                 name: str,
+        self.owner = owner
-                 contactData: str,
+        self.location = location
-                 apartamentAddress: str,
+        self.rooms = rooms
-                 roomsCount: int,
+        self.has_mamad = has_mamad
-                 whereIsBombShelter: str,
+        self.price = price
-                 costs: str
+        self.accepted_regions = accepted_regions
-                 ):
+        self.is_available = True
-        self.whereIsBombShelter = whereIsBombShelter
+
-        self.roomsCount = roomsCount
+    def __str__(self):
-        self.apartamentAddress = apartamentAddress
+        return (f"Apartment in {self.location}, {self.rooms} rooms, "
-        self.contactData = contactData
+                f"Safe Room: {'Yes' if self.has_mamad else 'No'}, "
-        self.name = name
+                f"Price: {self.price}, Available for: {', '.join(self.accepted_regions)}")
-        self.costs = costs

models/Apps.py

@@ -0,0 +1,119 @@
+from db_repository.BaseRepositories import BaseModelSchema
+from db_repository.volunteer_repository import *
+from db_repository.needy_repository import *
+from db_repository.apartment_repository import *
+from db_repository.schemas.apartment import *
+from db_repository.schemas.needy import *
+from db_repository.schemas.volunteer import *
+from models.Volunteer import Volunteer
+from models.Apartment import Apartment
+from models.Needy import Needy
+from threading import Lock
+
+mutex = Lock()
+
+
+class HousingApp:
+    def __init__(self, db_path: str = "app.db"):
+        self.db_path = db_path
+        self._init_repositories()
+
+    @property
+    def volunteers(self) -> List[Volunteer]:
+        return self.volunteer_repo.get_all()
+
+    @property
+    def apartaments(self) -> List[Apartment]:
+        return self.apartment_repo.get_all()
+
+    @property
+    def needys(self) -> List[Needy]:
+        return self.needy_repo.get_all()
+
+    def _init_repositories(self):
+        self.volunteer_repo = VolunteerRepository(mutex, self.db_path)
+        self.needy_repo = NeedyRepository(mutex, self.db_path)
+        self.apartment_repo = ApartmentRepository(mutex, self.db_path)
+
+    def register_volunteer(self, name: str, contact: str) -> BaseModelSchema | None:
+        volunteer = VolunteerSchema(name=name, contact=contact)
+        volunteer_id = self.volunteer_repo.create(volunteer)
+        return self.volunteer_repo.get(volunteer_id)
+
+    def register_evacuee(self, name: str, contact: str, region: str, how_much_people: int) -> NeedySchema:
+        evacuee = NeedySchema(
+            name=name,
+            contact=contact,
+            region=region,
+            how_much_peoples=how_much_people
+        )
+        evacuee_id = self.needy_repo.create(evacuee)
+        return self.needy_repo.get(evacuee_id)
+
+    def add_apartment(
+            self,
+            volunteer_name: str,
+            location: str,
+            rooms: int,
+            has_mamad: bool,
+            price: float,
+            regions: str
+    ) -> ApartmentSchema:
+        if not self.volunteer_repo.get_by_name(volunteer_name):
+            raise ValueError("Volunteer not found")
+
+        apartment = ApartmentSchema(
+            owner=volunteer_name,
+            location=location,
+            rooms=rooms,
+            has_mamad=str(has_mamad),
+            price=str(price),
+            accepted_regions=regions,
+            is_available=True
+        )
+        apartment_id = self.apartment_repo.create(apartment)
+
+        # Связываем квартиру с волонтером
+        self.volunteer_repo.add_apartment(volunteer_name, apartment_id)
+        return self.apartment_repo.get(apartment_id)
+
+    def search_apartments(self, evacuee_name: str) -> list[ApartmentSchema]:
+        evacuee = self.needy_repo.get_by_name(evacuee_name)
+        if not evacuee:
+            raise ValueError("Evacuee not found")
+
+        return self.apartment_repo.search_available(
+            evacuee.region,
+            1
+        )
+
+    def book_apartment(self, apartment_id: int, evacuee_name: str) -> bool:
+        apartment = self.apartment_repo.get(apartment_id)
+        evacuee = self.needy_repo.get_by_name(evacuee_name)
+
+        if not apartment or not evacuee:
+            return False
+
+        # Проверяем условия бронирования
+        if (
+                apartment.is_available and
+                (evacuee.region in apartment.accepted_regions or "all" in apartment.accepted_regions)
+        ):
+            updated_apartment = apartment.copy(update={"is_available": False})
+            self.apartment_repo.update(apartment_id, updated_apartment)
+            return True
+
+        return False
+
+    def get_volunteer_apartments(self, volunteer_id: int) -> list[ApartmentSchema]:
+        apartment_ids = self.volunteer_repo.get_apartments(volunteer_id)
+        return [self.apartment_repo.get(aid) for aid in apartment_ids]
+
+    def list_available_apartments(self) -> list[ApartmentSchema]:
+        return self.apartment_repo.get_all_available()
+
+    def list_all_evacuees(self) -> list[NeedySchema]:
+        return self.needy_repo.get_all()
+
+    def list_all_volunteers(self) -> list[VolunteerSchema]:
+        return self.volunteer_repo.get_all()

models/Needy.py

@@ -0,0 +1,8 @@
+from models.User import User
+
+
+class Needy(User):
+    def __init__(self, name: str, contact: str, region: str, how_much_peoples: int):
+        super().__init__(name, contact)
+        self.how_much_peoples = how_much_peoples
+        self.region = region

models/User.py

@@ -0,0 +1,4 @@
+class User:
+    def __init__(self, name, contact):
+        self.name = name
+        self.contact = contact

models/Volunteer.py

@@ -0,0 +1,7 @@
+from models.User import User
+
+
+class Volunteer(User):
+    def __init__(self, name, contact):
+        super().__init__(name, contact)
+        self.apartments = []