Spaces:
Runtime error
Runtime error
File size: 3,045 Bytes
8d21059 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | # database/posts.py
import logging
from database.connection import get_conn
logger = logging.getLogger(__name__)
def post_exists(message_id: int, channel_id: int) -> bool:
conn = get_conn()
try:
c = conn.cursor()
c.execute(
"SELECT id FROM posts WHERE message_id=? AND channel_id=?",
(message_id, channel_id)
)
return c.fetchone() is not None
except Exception as e:
logger.error(f"post_exists: {e}", exc_info=True)
return False
finally:
conn.close()
def insert_post(
message_id: int,
channel_id: int,
channel_username: str,
text_original: str,
text_normalized: str,
hashtags: str,
platform: str,
has_media: int,
media_type: str,
posted_at: str,
) -> bool:
conn = get_conn()
try:
c = conn.cursor()
c.execute("""
INSERT OR IGNORE INTO posts
(message_id, channel_id, channel_username, text_original,
text_normalized, hashtags, platform, has_media, media_type, posted_at)
VALUES (?,?,?,?,?,?,?,?,?,?)
""", (
message_id, channel_id, channel_username, text_original,
text_normalized, hashtags, platform, has_media, media_type, posted_at
))
conn.commit()
inserted = c.rowcount > 0
if inserted:
logger.info(
f"[DB] فُهرس msg_id={message_id} platform={platform} "
f"text='{text_original[:40]}'"
)
return inserted
except Exception as e:
logger.error(f"insert_post: {e}", exc_info=True)
return False
finally:
conn.close()
def delete_post(message_id: int, channel_id: int) -> bool:
conn = get_conn()
try:
c = conn.cursor()
c.execute(
"DELETE FROM posts WHERE message_id=? AND channel_id=?",
(message_id, channel_id)
)
conn.commit()
return c.rowcount > 0
except Exception as e:
logger.error(f"delete_post: {e}", exc_info=True)
return False
finally:
conn.close()
def get_total() -> int:
conn = get_conn()
try:
return conn.execute("SELECT COUNT(*) FROM posts").fetchone()[0]
except Exception:
return 0
finally:
conn.close()
def get_stats_by_platform() -> list[dict]:
conn = get_conn()
try:
c = conn.cursor()
c.execute("""
SELECT COALESCE(platform,'عام') AS platform, COUNT(*) AS cnt
FROM posts GROUP BY platform ORDER BY cnt DESC
""")
return [dict(r) for r in c.fetchall()]
except Exception:
return []
finally:
conn.close()
def rebuild_fts() -> bool:
conn = get_conn()
try:
conn.execute("INSERT INTO posts_fts(posts_fts) VALUES('rebuild')")
conn.commit()
return True
except Exception as e:
logger.error(f"rebuild_fts: {e}", exc_info=True)
return False
finally:
conn.close()
|