backup: 2026-04-09 17:52

Работы/ИРФМ/LR3/irfm_lr3.py

@@ -13,17 +13,33 @@ app = Flask(__name__)
 ALLOWED_FIELDS = ("lat", "lng", "title", "timeStamp", "town", "hour")
 
 
+def parse_positive_int(raw_value: str, field_name: str) -> tuple[int | None, str | None]:
+    """Преобразует строку в положительное целое число или возвращает текст ошибки."""
+    try:
+        value = int(raw_value)
+    except (TypeError, ValueError):
+        return None, f"{field_name} должен быть целым числом"
+
+    if value <= 0:
+        return None, f"{field_name} должен быть положительным целым числом"
+
+    return value, None
+
+
 def get_db_connection() -> sqlite3.Connection:
+    """Создает и возвращает подключение к SQLite с доступом к колонкам по имени."""
     conn = sqlite3.connect(DB_PATH)
     conn.row_factory = sqlite3.Row
     return conn
 
 
 def wants_wsdl() -> bool:
+    """Проверяет, запросил ли клиент XML/WSDL-формат через query-параметр format."""
     return request.args.get("format", "json").lower() == "wsdl"
 
 
 def dict_to_xml(payload: dict) -> bytes:
+    """Преобразует словарь (включая вложенные dict/list) в XML-документ."""
     root = Element("response")
 
     for key, value in payload.items():
@@ -53,6 +69,7 @@ def dict_to_xml(payload: dict) -> bytes:
 
 
 def format_response(payload: dict, status: int = 200) -> Response:
+    """Формирует ответ API: XML при format=wsdl, иначе JSON."""
     if wants_wsdl():
         xml_payload = dict_to_xml(payload)
         return Response(xml_payload, status=status, mimetype="application/wsdl+xml")
@@ -62,6 +79,7 @@ def format_response(payload: dict, status: int = 200) -> Response:
 
 
 def fetch_call_by_id(conn: sqlite3.Connection, call_id: int):
+    """Возвращает запись calls по rowid или None, если запись не найдена."""
     query = """
 		SELECT rowid AS id, lat, lng, title, timeStamp, town, hour
 		FROM calls
@@ -71,6 +89,7 @@ def fetch_call_by_id(conn: sqlite3.Connection, call_id: int):
 
 
 def parse_payload() -> tuple[dict, str | None]:
+    """Читает JSON-тело запроса, валидирует и оставляет только разрешенные поля."""
     if not request.is_json:
         return {}, "Body должен быть в формате JSON"
 
@@ -87,6 +106,7 @@ def parse_payload() -> tuple[dict, str | None]:
 
 @app.get("/api/calls")
 def list_calls() -> Response:
+    """Возвращает пагинированный список объектов calls."""
     page = max(request.args.get("page", default=1, type=int), 1)
     per_page = min(max(request.args.get("per_page", default=20, type=int), 1), 100)
     offset = (page - 1) * per_page
@@ -115,6 +135,10 @@ def list_calls() -> Response:
 
 @app.get("/api/calls/<int:call_id>")
 def get_call(call_id: int) -> Response:
+    """Возвращает один объект calls по его идентификатору."""
+    if call_id <= 0:
+        return format_response({"error": "id должен быть положительным целым числом"}, status=400)
+
     conn = get_db_connection()
     row = fetch_call_by_id(conn, call_id)
     conn.close()
@@ -127,6 +151,7 @@ def get_call(call_id: int) -> Response:
 
 @app.post("/api/calls")
 def create_call() -> Response:
+    """Создает новый объект calls из данных JSON-тела запроса."""
     payload, error = parse_payload()
     if error:
         return format_response({"error": error}, status=400)
@@ -157,6 +182,10 @@ def create_call() -> Response:
 
 @app.put("/api/calls/<int:call_id>")
 def update_call(call_id: int) -> Response:
+    """Обновляет существующий объект calls по id частичным набором полей."""
+    if call_id <= 0:
+        return format_response({"error": "id должен быть положительным целым числом"}, status=400)
+
     payload, error = parse_payload()
     if error:
         return format_response({"error": error}, status=400)
@@ -184,6 +213,10 @@ def update_call(call_id: int) -> Response:
 
 @app.delete("/api/calls/<int:call_id>")
 def delete_call(call_id: int) -> Response:
+    """Удаляет объект calls по id и возвращает статус операции."""
+    if call_id <= 0:
+        return format_response({"error": "id должен быть положительным целым числом"}, status=400)
+
     conn = get_db_connection()
     exists = fetch_call_by_id(conn, call_id)
     if exists is None:
@@ -199,6 +232,7 @@ def delete_call(call_id: int) -> Response:
 
 @app.get("/api/stats/hour/<int:hour>")
 def stats_by_hour(hour: int) -> Response:
+    """Возвращает количество обращений в таблице calls для заданного часа (0-23)."""
     if hour < 0 or hour > 23:
         return format_response({"error": "Час должен быть от 0 до 23"}, status=400)
 
@@ -215,8 +249,32 @@ def stats_by_hour(hour: int) -> Response:
     return format_response(payload)
 
 
+@app.route("/api/calls/<call_id>", methods=["GET", "PUT", "DELETE"])
+def invalid_call_id(call_id: str) -> Response:
+    """Возвращает понятную ошибку, если id в URL не является корректным числом."""
+    _, error = parse_positive_int(call_id, "id")
+    if error:
+        return format_response({"error": error}, status=400)
+    return format_response({"error": "Некорректный запрос"}, status=400)
+
+
+@app.get("/api/stats/hour/<hour>")
+def invalid_hour(hour: str) -> Response:
+    """Возвращает понятную ошибку, если hour в URL задан некорректно."""
+    try:
+        parsed_hour = int(hour)
+    except (TypeError, ValueError):
+        return format_response({"error": "Час должен быть целым числом"}, status=400)
+
+    if parsed_hour < 0 or parsed_hour > 23:
+        return format_response({"error": "Час должен быть от 0 до 23"}, status=400)
+
+    return format_response({"error": "Некорректный запрос"}, status=400)
+
+
 @app.get("/api/wsdl")
 def wsdl_description() -> Response:
+    """Отдает статическое WSDL-описание доступных операций сервиса."""
     wsdl = """<?xml version=\"1.0\" encoding=\"UTF-8\"?>
 <definitions name=\"CallsService\"
              targetNamespace=\"http://localhost:5000/calls\"