feat: alertas de picos, grafica de menciones por dia y estabilidad del traductor
- Backend: endpoint /api/entities/mentions (series por dia para N conceptos) - Backend: tabla alertas + escaner de picos automatico + endpoints (list/read/scan) - Frontend: pagina Evolucion con grafica SVG multi-concepto configurable - Frontend: pagina Alertas con badge de notificaciones en la barra - Traductor: memoria acotada (sub-lotes, batch 32, cuerpo max 12000 chars) - Traductor: prioridad baja en CPU (cpu_shares 512) y tope de 2 nucleos x worker - Escalado a 3 workers de traduccion (2x translator + translator-gpu)
This commit is contained in:
parent
eff656adea
commit
3991237685
11 changed files with 988 additions and 28 deletions
|
|
@ -76,6 +76,27 @@ func initDB() {
|
|||
ON CONFLICT (key) DO NOTHING
|
||||
`)
|
||||
|
||||
// Crear tabla de alertas si no existe
|
||||
_, err = db.GetPool().Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS alertas (
|
||||
id SERIAL PRIMARY KEY,
|
||||
valor VARCHAR(255) NOT NULL,
|
||||
tipo VARCHAR(32) NOT NULL,
|
||||
periodo DATE NOT NULL,
|
||||
hits INT NOT NULL,
|
||||
baseline DOUBLE PRECISION NOT NULL,
|
||||
ratio DOUBLE PRECISION NOT NULL,
|
||||
status VARCHAR(16) NOT NULL DEFAULT 'nueva',
|
||||
created_at TIMESTAMP DEFAULT NOW(),
|
||||
UNIQUE(valor, tipo, periodo)
|
||||
)
|
||||
`)
|
||||
if err != nil {
|
||||
log.Printf("Warning: Could not create alertas table: %v", err)
|
||||
} else {
|
||||
log.Println("Table alertas ready")
|
||||
}
|
||||
|
||||
// Crear tabla de remote_workers si no existe
|
||||
_, err = db.GetPool().Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS remote_workers (
|
||||
|
|
@ -175,6 +196,11 @@ func main() {
|
|||
|
||||
api.GET("/entities", handlers.GetEntities)
|
||||
api.GET("/entities/news", handlers.GetEntityNews)
|
||||
api.GET("/entities/mentions", handlers.GetEntityMentions)
|
||||
|
||||
api.GET("/alerts", handlers.GetAlertas)
|
||||
api.POST("/alerts/:id/read", middleware.AuthRequired(), handlers.MarkAlertaRead)
|
||||
api.POST("/alerts/read-all", middleware.AuthRequired(), handlers.MarkAllAlertasRead)
|
||||
|
||||
api.GET("/stats", handlers.GetStats)
|
||||
|
||||
|
|
@ -195,6 +221,7 @@ func main() {
|
|||
admin.POST("/users/:id/promote", handlers.PromoteUser)
|
||||
admin.POST("/users/:id/demote", handlers.DemoteUser)
|
||||
admin.POST("/reset-db", handlers.ResetDatabase)
|
||||
admin.POST("/alerts/scan", handlers.ScanAlertasAdmin)
|
||||
admin.GET("/workers/status", handlers.GetWorkerStatus)
|
||||
admin.GET("/workers/stats", handlers.GetTranslationStats)
|
||||
admin.POST("/workers/config", handlers.SetWorkerConfig)
|
||||
|
|
@ -231,6 +258,7 @@ func main() {
|
|||
}()
|
||||
|
||||
handlers.StartJobAssigner()
|
||||
handlers.StartAlertScanner()
|
||||
|
||||
quit := make(chan os.Signal, 1)
|
||||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||||
|
|
|
|||
217
backend/internal/handlers/alerts.go
Normal file
217
backend/internal/handlers/alerts.go
Normal file
|
|
@ -0,0 +1,217 @@
|
|||
package handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/rss2/backend/internal/db"
|
||||
"github.com/rss2/backend/internal/models"
|
||||
)
|
||||
|
||||
type Alerta struct {
|
||||
ID int64 `json:"id"`
|
||||
Valor string `json:"valor"`
|
||||
Tipo string `json:"tipo"`
|
||||
Periodo string `json:"periodo"`
|
||||
Hits int `json:"hits"`
|
||||
Baseline float64 `json:"baseline"`
|
||||
Ratio float64 `json:"ratio"`
|
||||
Status string `json:"status"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
func GetAlertas(c *gin.Context) {
|
||||
status := c.Query("status")
|
||||
limitStr := c.DefaultQuery("limit", "100")
|
||||
|
||||
limit, _ := strconv.Atoi(limitStr)
|
||||
if limit < 1 || limit > 500 {
|
||||
limit = 100
|
||||
}
|
||||
|
||||
query := `
|
||||
SELECT id, valor, tipo, periodo::text, hits, baseline, ratio, status, to_char(created_at, 'YYYY-MM-DD HH24:MI') AS created_at
|
||||
FROM alertas
|
||||
`
|
||||
args := []interface{}{}
|
||||
if status != "" {
|
||||
query += fmt.Sprintf(" WHERE status = $%d", len(args)+1)
|
||||
args = append(args, status)
|
||||
}
|
||||
query += fmt.Sprintf(" ORDER BY periodo DESC, ratio DESC, id DESC LIMIT $%d", len(args)+1)
|
||||
args = append(args, limit)
|
||||
|
||||
rows, err := db.GetPool().Query(c.Request.Context(), query, args...)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.ErrorResponse{Error: "Failed to get alerts", Message: err.Error()})
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
alertas := []Alerta{}
|
||||
for rows.Next() {
|
||||
var a Alerta
|
||||
if err := rows.Scan(&a.ID, &a.Valor, &a.Tipo, &a.Periodo, &a.Hits, &a.Baseline, &a.Ratio, &a.Status, &a.CreatedAt); err != nil {
|
||||
continue
|
||||
}
|
||||
alertas = append(alertas, a)
|
||||
}
|
||||
|
||||
var nuevas int
|
||||
db.GetPool().QueryRow(c.Request.Context(), "SELECT COUNT(*)::int FROM alertas WHERE status = 'nueva'").Scan(&nuevas)
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"alertas": alertas,
|
||||
"total": len(alertas),
|
||||
"nuevas": nuevas,
|
||||
})
|
||||
}
|
||||
|
||||
func MarkAlertaRead(c *gin.Context) {
|
||||
idStr := c.Param("id")
|
||||
id, err := strconv.ParseInt(idStr, 10, 64)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, models.ErrorResponse{Error: "Invalid alert id"})
|
||||
return
|
||||
}
|
||||
|
||||
_, err = db.GetPool().Exec(c.Request.Context(), "UPDATE alertas SET status = 'leida' WHERE id = $1", id)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.ErrorResponse{Error: "Failed to update alert", Message: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
func MarkAllAlertasRead(c *gin.Context) {
|
||||
_, err := db.GetPool().Exec(c.Request.Context(), "UPDATE alertas SET status = 'leida' WHERE status = 'nueva'")
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.ErrorResponse{Error: "Failed to update alerts", Message: err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
func ScanAlertasAdmin(c *gin.Context) {
|
||||
inserted, err := RunAlertScan(c.Request.Context())
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.ErrorResponse{Error: "Scan failed", Message: err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true, "new_alerts": inserted})
|
||||
}
|
||||
|
||||
// RunAlertScan busca conceptos (persona, lugar, organizacion, tema) cuya
|
||||
// actividad de hoy supera significativamente su media de los últimos días.
|
||||
func RunAlertScan(ctx context.Context) (int, error) {
|
||||
minHits := 3
|
||||
minRatio := 5.0
|
||||
if v := os.Getenv("ALERTS_MIN_HITS"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
minHits = n
|
||||
}
|
||||
}
|
||||
if v := os.Getenv("ALERTS_MIN_RATIO"); v != "" {
|
||||
if f, err := strconv.ParseFloat(v, 64); err == nil && f > 0 {
|
||||
minRatio = f
|
||||
}
|
||||
}
|
||||
|
||||
query := `
|
||||
WITH daily AS (
|
||||
SELECT t.id AS tag_id, t.valor, t.tipo, n.fecha::date AS dia,
|
||||
COUNT(DISTINCT n.id) AS cnt
|
||||
FROM tags_noticia tn
|
||||
JOIN tags t ON tn.tag_id = t.id
|
||||
JOIN traducciones tr ON tn.traduccion_id = tr.id
|
||||
JOIN noticias n ON tr.noticia_id = n.id
|
||||
WHERE n.fecha >= CURRENT_DATE - 8
|
||||
AND t.tipo IN ('persona', 'lugar', 'organizacion', 'tema')
|
||||
GROUP BY t.id, t.valor, t.tipo, n.fecha::date
|
||||
),
|
||||
baseline AS (
|
||||
SELECT tag_id, valor, tipo,
|
||||
AVG(cnt)::float AS baseline,
|
||||
MAX(cnt)::int AS max_prev
|
||||
FROM daily
|
||||
WHERE dia < CURRENT_DATE
|
||||
GROUP BY tag_id, valor, tipo
|
||||
),
|
||||
today AS (
|
||||
SELECT tag_id, valor, tipo, SUM(cnt)::int AS hits
|
||||
FROM daily
|
||||
WHERE dia = CURRENT_DATE
|
||||
GROUP BY tag_id, valor, tipo
|
||||
)
|
||||
SELECT t.valor, t.tipo, t.hits, b.baseline,
|
||||
(t.hits::float / NULLIF(b.baseline, 0)) AS ratio
|
||||
FROM today t
|
||||
JOIN baseline b USING (tag_id)
|
||||
WHERE t.hits >= $1
|
||||
AND b.baseline >= 0.5
|
||||
AND (t.hits::float / NULLIF(b.baseline, 0)) >= $2
|
||||
`
|
||||
|
||||
rows, err := db.GetPool().Query(ctx, query, minHits, minRatio)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
inserted := 0
|
||||
for rows.Next() {
|
||||
var valor, tipo string
|
||||
var hits int
|
||||
var baseline, ratio float64
|
||||
if err := rows.Scan(&valor, &tipo, &hits, &baseline, &ratio); err != nil {
|
||||
continue
|
||||
}
|
||||
res, err := db.GetPool().Exec(ctx, `
|
||||
INSERT INTO alertas (valor, tipo, periodo, hits, baseline, ratio)
|
||||
VALUES ($1, $2, CURRENT_DATE, $3, $4, $5)
|
||||
ON CONFLICT (valor, tipo, periodo) DO NOTHING
|
||||
`, valor, tipo, hits, baseline, ratio)
|
||||
if err != nil {
|
||||
log.Printf("alerts: insert failed for %s (%s): %v", valor, tipo, err)
|
||||
continue
|
||||
}
|
||||
if n := res.RowsAffected(); n > 0 {
|
||||
inserted++
|
||||
}
|
||||
}
|
||||
|
||||
return inserted, nil
|
||||
}
|
||||
|
||||
// StartAlertScanner lanza una goroutine que revisa picos de actividad
|
||||
// periódicamente (cada ALERTS_SCAN_INTERVAL_MIN, por defecto 120 min).
|
||||
func StartAlertScanner() {
|
||||
interval := 120
|
||||
if v := os.Getenv("ALERTS_SCAN_INTERVAL_MIN"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
interval = n
|
||||
}
|
||||
}
|
||||
|
||||
go func() {
|
||||
time.Sleep(45 * time.Second)
|
||||
for {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
inserted, err := RunAlertScan(ctx)
|
||||
cancel()
|
||||
if err != nil {
|
||||
log.Printf("alerts: scan error: %v", err)
|
||||
} else {
|
||||
log.Printf("alerts: scan finished, %d new alerts", inserted)
|
||||
}
|
||||
time.Sleep(time.Duration(interval) * time.Minute)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
@ -1,9 +1,11 @@
|
|||
package handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
|
@ -484,3 +486,140 @@ func GetEntities(c *gin.Context) {
|
|||
TotalPages: totalPages,
|
||||
})
|
||||
}
|
||||
|
||||
type MentionPoint struct {
|
||||
Fecha string `json:"fecha"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
type MentionSeries struct {
|
||||
Valor string `json:"valor"`
|
||||
Tipo string `json:"tipo"`
|
||||
Count int `json:"count"`
|
||||
Data []MentionPoint `json:"data"`
|
||||
}
|
||||
|
||||
type MentionsResponse struct {
|
||||
Days int `json:"days"`
|
||||
Series []MentionSeries `json:"series"`
|
||||
}
|
||||
|
||||
func GetEntityMentions(c *gin.Context) {
|
||||
valuesRaw := c.DefaultQuery("values", "")
|
||||
days := 30
|
||||
if v := c.Query("days"); v != "" {
|
||||
if d, err := strconv.Atoi(v); err == nil && d >= 1 && d <= 365 {
|
||||
days = d
|
||||
}
|
||||
}
|
||||
|
||||
var values []string
|
||||
for _, v := range strings.Split(valuesRaw, ",") {
|
||||
v = strings.TrimSpace(v)
|
||||
if v != "" {
|
||||
values = append(values, v)
|
||||
}
|
||||
}
|
||||
if len(values) == 0 {
|
||||
c.JSON(http.StatusBadRequest, models.ErrorResponse{Error: "values is required (comma separated)"})
|
||||
return
|
||||
}
|
||||
if len(values) > 20 {
|
||||
c.JSON(http.StatusBadRequest, models.ErrorResponse{Error: "Too many values (max 20)"})
|
||||
return
|
||||
}
|
||||
|
||||
start := time.Now().AddDate(0, 0, -(days - 1)).Truncate(24 * time.Hour)
|
||||
|
||||
series := make([]MentionSeries, 0, len(values))
|
||||
for _, valor := range values {
|
||||
s, err := buildMentionSeries(c.Request.Context(), valor, days)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.ErrorResponse{Error: "Failed to get mentions", Message: err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
m := map[string]int{}
|
||||
for _, p := range s.Data {
|
||||
m[p.Fecha] = p.Count
|
||||
}
|
||||
data := make([]MentionPoint, 0, days)
|
||||
for i := 0; i < days; i++ {
|
||||
fecha := start.AddDate(0, 0, i).Format("2006-01-02")
|
||||
data = append(data, MentionPoint{Fecha: fecha, Count: m[fecha]})
|
||||
}
|
||||
s.Data = data
|
||||
s.Count = 0
|
||||
for _, p := range s.Data {
|
||||
s.Count += p.Count
|
||||
}
|
||||
if s.Tipo == "" {
|
||||
s.Tipo = "desconocido"
|
||||
}
|
||||
series = append(series, *s)
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, MentionsResponse{Days: days, Series: series})
|
||||
}
|
||||
|
||||
func buildMentionSeries(ctx context.Context, valor string, days int) (*MentionSeries, error) {
|
||||
s := &MentionSeries{Valor: valor}
|
||||
|
||||
var tagIDs []int64
|
||||
rows, err := db.GetPool().Query(ctx, `
|
||||
SELECT DISTINCT t.id, t.tipo
|
||||
FROM tags t
|
||||
LEFT JOIN entity_aliases ea ON ea.tipo = t.tipo AND LOWER(ea.alias) = LOWER(t.valor)
|
||||
WHERE LOWER(t.valor) = LOWER($1)
|
||||
OR LOWER(COALESCE(ea.canonical_name, t.valor)) = LOWER($1)
|
||||
`, valor)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
var tipo string
|
||||
if err := rows.Scan(&id, &tipo); err != nil {
|
||||
continue
|
||||
}
|
||||
tagIDs = append(tagIDs, id)
|
||||
if s.Tipo == "" {
|
||||
s.Tipo = tipo
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
if len(tagIDs) == 0 {
|
||||
s.Data = []MentionPoint{}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
drows, err := db.GetPool().Query(ctx, `
|
||||
SELECT n.fecha::date::text AS dia, COUNT(DISTINCT n.id) AS cnt
|
||||
FROM tags_noticia tn
|
||||
JOIN tags t ON tn.tag_id = t.id
|
||||
JOIN traducciones tr ON tn.traduccion_id = tr.id
|
||||
JOIN noticias n ON tr.noticia_id = n.id
|
||||
WHERE t.id = ANY($1) AND n.fecha >= CURRENT_DATE - ($2::int - 1)
|
||||
GROUP BY n.fecha::date
|
||||
ORDER BY n.fecha::date
|
||||
`, tagIDs, days)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer drows.Close()
|
||||
|
||||
for drows.Next() {
|
||||
var fecha string
|
||||
var cnt int
|
||||
if err := drows.Scan(&fecha, &cnt); err != nil {
|
||||
continue
|
||||
}
|
||||
s.Data = append(s.Data, MentionPoint{Fecha: fecha, Count: cnt})
|
||||
}
|
||||
if s.Data == nil {
|
||||
s.Data = []MentionPoint{}
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue