import psycopg2
from flask import Flask, g, render_template_string, request
from psycopg2.extras import RealDictCursor
app = Flask(__name__)
DB_CONFIG = {
"host": "postgres.openclaw.svc.cluster.local",
"port": 5432,
"user": "edisclosure",
"password": "veJEFrqvLLtd",
"database": "edisclosure",
}
def get_db():
db = getattr(g, "_database", None)
if db is None:
db = g._database = psycopg2.connect(**DB_CONFIG, cursor_factory=RealDictCursor)
return db
@app.teardown_appcontext
def close_connection(exception):
db = getattr(g, "_database", None)
if db is not None:
db.close()
HTML = """
eDisk API
{{ content | safe }}
"""
HOME_HTML = """
eDisclosure Events API
REST API для доступа к событиям раскрытия информации компаниями.
Endpoints
GET /events
Получить список событий с фильтрацией, сортировкой и пагинацией.
Параметры:
| Параметр | Тип | Описание |
|---|
secid | string | Фильтр по тикеру ценной бумаги (например, GAZP, SBER) |
inn | string | Фильтр по ИНН компании |
q | string | Поиск по названию события и названию компании (substring) |
text | string | Поиск по полному тексту события (substring) |
sort | string | Сортировка по дате: desc (новые первыми) или asc (старые первыми) |
page | int | Номер страницы (по умолчанию: 1) |
size | int | Размер страницы (по умолчанию: 50, макс: 100) |
Примеры запросов:
/events?secid=GAZP&sort=desc
/events?inn=7736050003&sort=asc&page=2
/events?q=дивиденд&text=выплата
/events?sort=desc&size=100
GET /event/<event_id>
Получить детали конкретного события по его ID.
Пример:
/event/HRpA9UHzzUu9GQuXbnu4pA-B-B
Статистика базы данных
- Всего событий: {{ events_count }}
- Всего компаний: {{ companies_count }}
"""
EVENTS_HTML = """
Events
Всего: {{ total }}
| Дата | Компания | Событие | SECID |
|---|
{% for row in rows %}
| {{ row.event_date }} |
{{ row.company_name or '' }} |
{{ row.event_name or '' }} |
{{ row.secid or '' }} |
{% endfor %}
"""
EVENT_DETAIL_HTML = """
{{ event.event_name or 'Event Details' }}
Дата: {{ event.event_date }} |
Компания: {{ event.company_name or '' }} |
INN: {{ event.inn or '' }} |
SECID: {{ event.secid or '' }}
Полный текст:
{{ event.text or 'Нет данных' }}
← Назад к списку
"""
@app.route("/")
def index():
db = get_db()
cur = db.cursor()
cur.execute("SELECT COUNT(*) as cnt FROM events_all")
events_count = cur.fetchone()["cnt"]
cur.execute("SELECT COUNT(*) as cnt FROM companies")
companies_count = cur.fetchone()["cnt"]
cur.close()
return render_template_string(
HTML.replace("{{ content | safe }}", HOME_HTML),
active="home",
events_count=events_count,
companies_count=companies_count,
)
@app.route("/events")
def events():
db = get_db()
cur = db.cursor()
search = request.args.get("q", "")
inn = request.args.get("inn", "")
secid = request.args.get("secid", "")
text_search = request.args.get("text", "")
sort = request.args.get("sort", "desc")
page = int(request.args.get("page", 1))
size = min(int(request.args.get("size", 50)), 200)
where, params = [], []
if search:
where.append("(event_name LIKE %s OR company_name LIKE %s)")
params.extend([f"%{search}%", f"%{search}%"])
if inn:
where.append("inn = %s")
params.append(inn)
if secid:
where.append("secid = %s")
params.append(secid)
if text_search:
where.append("text LIKE %s")
params.append(f"%{text_search}%")
order = (
"TO_DATE(event_date, 'DD.MM.YYYY') DESC"
if sort == "desc"
else "TO_DATE(event_date, 'DD.MM.YYYY') ASC"
)
if where:
where_str = "WHERE " + " AND ".join(where)
cur.execute(f"SELECT COUNT(*) as cnt FROM events_all {where_str}", params)
total = cur.fetchone()["cnt"]
cur.execute(
f"SELECT event_id, event_date, event_name, company_name, inn, secid "
f"FROM events_all {where_str} ORDER BY {order} LIMIT {size} OFFSET {(page - 1) * size}",
params,
)
else:
cur.execute("SELECT COUNT(*) as cnt FROM events_all")
total = cur.fetchone()["cnt"]
cur.execute(
f"SELECT event_id, event_date, event_name, company_name, inn, secid "
f"FROM events_all ORDER BY {order} LIMIT {size} OFFSET {(page - 1) * size}"
)
rows = cur.fetchall()
cur.close()
pages = (total + size - 1) // size if total > 0 else 1
return render_template_string(
HTML.replace("{{ content | safe }}", EVENTS_HTML),
active="events",
rows=rows,
total=total,
page=page,
pages=pages,
search=search,
secid=secid,
inn=inn,
sort=sort,
size=size,
)
@app.route("/event/")
def event_detail(event_id):
db = get_db()
cur = db.cursor()
cur.execute(
"SELECT event_id, event_date, event_name, company_name, inn, secid, text "
"FROM events_all WHERE event_id = %s",
(event_id,),
)
event = cur.fetchone()
cur.close()
if event is None:
return render_template_string(
HTML.replace(
"{{ content | safe }}",
"Событие не найдено
← Назад",
),
active="events",
)
return render_template_string(
HTML.replace("{{ content | safe }}", EVENT_DETAIL_HTML),
active="events",
event=event,
)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=80)