Bulk rename/move files to '8' directories
Perform a bulk rename/move of many files and folders: paths previously prefixed with '01', '02', '03', '04', etc. were relocated to equivalent paths prefixed with '8'. Changes are path-only (R100 renames) and do not modify file contents; this reorganizes the repository structure.
This commit is contained in:
Dmitry
@@ -0,0 +1,168 @@
|
||||
import json
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
BASE_URL = "http://127.0.0.1:5000"
|
||||
|
||||
|
||||
def ask_format() -> str:
|
||||
"""Запрашивает желаемый формат ответа и возвращает json или wsdl."""
|
||||
raw = input("Формат ответа (json/wsdl) [json]: ").strip().lower()
|
||||
return raw if raw in {"json", "wsdl"} else "json"
|
||||
|
||||
|
||||
def print_response(resp: requests.Response) -> None:
|
||||
"""Печатает HTTP-статус и тело ответа в JSON или сыром виде."""
|
||||
print(f"\nHTTP {resp.status_code}")
|
||||
try:
|
||||
print(json.dumps(resp.json(), ensure_ascii=False, indent=2))
|
||||
except ValueError:
|
||||
print(resp.text)
|
||||
|
||||
|
||||
def input_payload() -> dict:
|
||||
"""Собирает поля объекта из консоли и возвращает payload для POST/PUT."""
|
||||
print("Оставьте поле пустым, если не хотите его передавать")
|
||||
payload = {}
|
||||
|
||||
lat = input("lat: ").strip()
|
||||
if lat:
|
||||
payload["lat"] = float(lat)
|
||||
|
||||
lng = input("lng: ").strip()
|
||||
if lng:
|
||||
payload["lng"] = float(lng)
|
||||
|
||||
title = input("title: ").strip()
|
||||
if title:
|
||||
payload["title"] = title
|
||||
|
||||
timestamp = input("timeStamp (YYYY-MM-DD HH:MM:SS): ").strip()
|
||||
if timestamp:
|
||||
payload["timeStamp"] = timestamp
|
||||
|
||||
town = input("town: ").strip()
|
||||
if town:
|
||||
payload["town"] = town
|
||||
|
||||
hour = input("hour (0-23): ").strip()
|
||||
if hour:
|
||||
payload["hour"] = int(hour)
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
def list_calls() -> None:
|
||||
"""Запрашивает и выводит пагинированный список объектов calls."""
|
||||
page = input("page [1]: ").strip() or "1"
|
||||
per_page = input("per_page [20]: ").strip() or "20"
|
||||
fmt = ask_format()
|
||||
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/calls",
|
||||
params={"page": page, "per_page": per_page, "format": fmt},
|
||||
timeout=30,
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def get_call() -> None:
|
||||
"""Запрашивает и выводит один объект calls по ID."""
|
||||
call_id = input("ID объекта: ").strip()
|
||||
fmt = ask_format()
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/calls/{call_id}", params={"format": fmt}, timeout=30
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def create_call() -> None:
|
||||
"""Создает новый объект calls из введенных пользователем данных."""
|
||||
payload = input_payload()
|
||||
fmt = ask_format()
|
||||
resp = requests.post(
|
||||
f"{BASE_URL}/api/calls", params={"format": fmt}, json=payload, timeout=30
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def update_call() -> None:
|
||||
"""Обновляет существующий объект calls по ID введенными полями."""
|
||||
call_id = input("ID объекта: ").strip()
|
||||
payload = input_payload()
|
||||
fmt = ask_format()
|
||||
resp = requests.put(
|
||||
f"{BASE_URL}/api/calls/{call_id}",
|
||||
params={"format": fmt},
|
||||
json=payload,
|
||||
timeout=30,
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def delete_call() -> None:
|
||||
"""Удаляет объект calls по ID и выводит результат операции."""
|
||||
call_id = input("ID объекта: ").strip()
|
||||
fmt = ask_format()
|
||||
resp = requests.delete(
|
||||
f"{BASE_URL}/api/calls/{call_id}", params={"format": fmt}, timeout=30
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def stats_by_hour() -> None:
|
||||
"""Запрашивает статистику количества обращений по выбранному часу."""
|
||||
hour = input("Час (0-23): ").strip()
|
||||
fmt = ask_format()
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/stats/hour/{hour}", params={"format": fmt}, timeout=30
|
||||
)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def show_wsdl() -> None:
|
||||
"""Получает и выводит WSDL-описание API."""
|
||||
resp = requests.get(f"{BASE_URL}/api/wsdl", timeout=30)
|
||||
print_response(resp)
|
||||
|
||||
|
||||
def menu() -> None:
|
||||
"""Запускает интерактивное меню CLI и обрабатывает выбор пользователя."""
|
||||
actions = {
|
||||
"1": ("Список объектов", list_calls),
|
||||
"2": ("Один объект по ID", get_call),
|
||||
"3": ("Добавить объект", create_call),
|
||||
"4": ("Изменить объект", update_call),
|
||||
"5": ("Удалить объект", delete_call),
|
||||
"6": ("Общее число обращений в час", stats_by_hour),
|
||||
"7": ("Показать WSDL", show_wsdl),
|
||||
"0": ("Выход", None),
|
||||
}
|
||||
|
||||
while True:
|
||||
print("\n=== CLI для API LR3 ===")
|
||||
for key, (name, _) in actions.items():
|
||||
print(f"{key}. {name}")
|
||||
|
||||
choice = input("Выберите метод: ").strip()
|
||||
action = actions.get(choice)
|
||||
|
||||
if not action:
|
||||
print("Неизвестный пункт меню")
|
||||
continue
|
||||
|
||||
if choice == "0":
|
||||
print("Завершение работы")
|
||||
break
|
||||
|
||||
try:
|
||||
action[1]()
|
||||
except requests.RequestException as exc:
|
||||
print(f"Ошибка запроса: {exc}")
|
||||
except ValueError as exc:
|
||||
print(f"Ошибка ввода: {exc}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
menu()
|
||||
@@ -0,0 +1,300 @@
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from xml.etree.ElementTree import Element, SubElement, tostring
|
||||
|
||||
from flask import Flask, Response, jsonify, request
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent.parent
|
||||
DB_PATH = BASE_DIR / "data" / "calls.db"
|
||||
|
||||
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():
|
||||
if isinstance(value, dict):
|
||||
obj_node = SubElement(root, str(key))
|
||||
for child_key, child_value in value.items():
|
||||
child = SubElement(obj_node, str(child_key))
|
||||
child.text = "" if child_value is None else str(child_value)
|
||||
continue
|
||||
|
||||
if isinstance(value, list):
|
||||
list_node = SubElement(root, str(key))
|
||||
for item in value:
|
||||
item_node = SubElement(list_node, "item")
|
||||
if isinstance(item, dict):
|
||||
for item_key, item_value in item.items():
|
||||
child = SubElement(item_node, str(item_key))
|
||||
child.text = "" if item_value is None else str(item_value)
|
||||
else:
|
||||
item_node.text = "" if item is None else str(item)
|
||||
continue
|
||||
|
||||
node = SubElement(root, str(key))
|
||||
node.text = "" if value is None else str(value)
|
||||
|
||||
return tostring(root, encoding="utf-8", xml_declaration=True)
|
||||
|
||||
|
||||
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")
|
||||
resp = jsonify(payload)
|
||||
resp.status_code = status
|
||||
return resp
|
||||
|
||||
|
||||
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
|
||||
WHERE rowid = ?
|
||||
"""
|
||||
return conn.execute(query, (call_id,)).fetchone()
|
||||
|
||||
|
||||
def parse_payload() -> tuple[dict, str | None]:
|
||||
"""Читает JSON-тело запроса, валидирует и оставляет только разрешенные поля."""
|
||||
if not request.is_json:
|
||||
return {}, "Body должен быть в формате JSON"
|
||||
|
||||
payload = request.get_json(silent=True)
|
||||
if not isinstance(payload, dict):
|
||||
return {}, "Некорректный JSON-объект"
|
||||
|
||||
filtered = {k: payload[k] for k in ALLOWED_FIELDS if k in payload}
|
||||
if not filtered:
|
||||
return {}, "Не переданы поля для записи"
|
||||
|
||||
return filtered, 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
|
||||
|
||||
conn = get_db_connection()
|
||||
total = conn.execute("SELECT COUNT(*) AS total FROM calls").fetchone()["total"]
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT rowid AS id, lat, lng, title, timeStamp, town, hour
|
||||
FROM calls
|
||||
ORDER BY rowid
|
||||
LIMIT ? OFFSET ?
|
||||
""",
|
||||
(per_page, offset),
|
||||
).fetchall()
|
||||
conn.close()
|
||||
|
||||
payload = {
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"total": total,
|
||||
"items": [dict(row) for row in rows],
|
||||
}
|
||||
return format_response(payload)
|
||||
|
||||
|
||||
@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()
|
||||
|
||||
if row is None:
|
||||
return format_response({"error": "Объект не найден"}, status=404)
|
||||
|
||||
return format_response({"item": dict(row)})
|
||||
|
||||
|
||||
@app.post("/api/calls")
|
||||
def create_call() -> Response:
|
||||
"""Создает новый объект calls из данных JSON-тела запроса."""
|
||||
payload, error = parse_payload()
|
||||
if error:
|
||||
return format_response({"error": error}, status=400)
|
||||
|
||||
columns = ", ".join(payload.keys())
|
||||
placeholders = ", ".join(["?"] * len(payload))
|
||||
|
||||
conn = get_db_connection()
|
||||
cur = conn.execute(
|
||||
f"INSERT INTO calls ({columns}) VALUES ({placeholders})",
|
||||
tuple(payload.values()),
|
||||
)
|
||||
conn.commit()
|
||||
created_id = cur.lastrowid
|
||||
if created_id is None:
|
||||
conn.close()
|
||||
return format_response({"error": "Не удалось создать объект"}, status=500)
|
||||
|
||||
created_row = fetch_call_by_id(conn, int(created_id))
|
||||
conn.close()
|
||||
if created_row is None:
|
||||
return format_response(
|
||||
{"error": "Не удалось получить созданный объект"}, status=500
|
||||
)
|
||||
|
||||
return format_response({"item": dict(created_row)}, status=201)
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
assignments = ", ".join([f"{key} = ?" for key in payload.keys()])
|
||||
values = tuple(payload.values()) + (call_id,)
|
||||
|
||||
conn = get_db_connection()
|
||||
exists = fetch_call_by_id(conn, call_id)
|
||||
if exists is None:
|
||||
conn.close()
|
||||
return format_response({"error": "Объект не найден"}, status=404)
|
||||
|
||||
conn.execute(f"UPDATE calls SET {assignments} WHERE rowid = ?", values)
|
||||
conn.commit()
|
||||
updated_row = fetch_call_by_id(conn, call_id)
|
||||
conn.close()
|
||||
if updated_row is None:
|
||||
return format_response(
|
||||
{"error": "Не удалось получить обновленный объект"}, status=500
|
||||
)
|
||||
|
||||
return format_response({"item": dict(updated_row)})
|
||||
|
||||
|
||||
@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:
|
||||
conn.close()
|
||||
return format_response({"error": "Объект не найден"}, status=404)
|
||||
|
||||
conn.execute("DELETE FROM calls WHERE rowid = ?", (call_id,))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
return format_response({"message": f"Объект {call_id} удален"})
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
conn = get_db_connection()
|
||||
total = conn.execute(
|
||||
"SELECT COUNT(*) AS total FROM calls WHERE hour = ?", (hour,)
|
||||
).fetchone()["total"]
|
||||
conn.close()
|
||||
|
||||
payload = {
|
||||
"hour": hour,
|
||||
"total_calls": total,
|
||||
}
|
||||
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\"
|
||||
xmlns:tns=\"http://localhost:5000/calls\"
|
||||
xmlns=\"http://schemas.xmlsoap.org/wsdl/\">
|
||||
<service name=\"CallsService\">
|
||||
<documentation>REST API для сущности calls из LR2.</documentation>
|
||||
</service>
|
||||
<portType name=\"CallsPortType\">
|
||||
<operation name=\"listCalls\" />
|
||||
<operation name=\"getCall\" />
|
||||
<operation name=\"createCall\" />
|
||||
<operation name=\"updateCall\" />
|
||||
<operation name=\"deleteCall\" />
|
||||
<operation name=\"statsByHour\" />
|
||||
</portType>
|
||||
</definitions>
|
||||
"""
|
||||
return Response(wsdl, mimetype="application/wsdl+xml")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(debug=True)
|
||||
Binary file not shown.
Reference in New Issue
Block a user