sajad-dev
2c42ebe01c
- Removed deprecated user_info files and paths from configuration. - Added user soil data integration in chat context to improve response accuracy. - Updated build_rag_context and chat_rag_stream functions to include sensor_uuid for user-specific data retrieval. - Enhanced load_sources function to load user data from the database. - Implemented filtering in search_with_query and QdrantVectorStore to isolate user data based on sensor_uuid. - Introduced Celery Beat schedule for periodic user data ingestion.
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""
|
|
ساخت دیتای خاک کاربر از sensor_data و soil_data — Schema-agnostic
|
|
هر سنسور = یک کاربر. شناسایی با uuid_sensor.
|
|
|
|
مدلهای Django داخل توابع import میشوند تا از AppRegistryNotReady جلوگیری شود.
|
|
"""
|
|
from django.db.models import Model
|
|
|
|
|
|
# فیلدهایی که در متن embed نباید بیایند (شناسهها، رابطهها)
|
|
EXCLUDE_FIELD_NAMES = {"id", "created_at", "updated_at", "task_id", "recorded_at"}
|
|
|
|
|
|
def _model_to_data_fields(instance: Model, exclude: set[str] | None = None) -> dict:
|
|
"""
|
|
استخراج فیلدهای داده از یک instance با استفاده از introspection.
|
|
تغییرات بعدی در مدل باعث شکستن نمیشود.
|
|
"""
|
|
exclude = exclude or set()
|
|
out: dict = {}
|
|
for f in instance._meta.get_fields():
|
|
if f.many_to_many or f.one_to_many or f.one_to_one and f.auto_created:
|
|
continue
|
|
if f.name in exclude or f.name in EXCLUDE_FIELD_NAMES:
|
|
continue
|
|
if hasattr(f, "related_model") and f.related_model:
|
|
continue # FK
|
|
try:
|
|
val = getattr(instance, f.name, None)
|
|
if val is not None:
|
|
out[f.name] = val
|
|
except Exception:
|
|
pass
|
|
return out
|
|
|
|
|
|
def build_user_soil_text(sensor_uuid: str) -> str | None:
|
|
"""
|
|
ساخت متن قابل embed برای یک سنسور (کاربر).
|
|
از SensorData → SoilLocation → SoilDepthData خوانده میشود.
|
|
|
|
Returns:
|
|
متن متنی قابل چانک، یا None اگر سنسور یافت نشد.
|
|
"""
|
|
from sensor_data.models import SensorData
|
|
from soil_data.models import SoilDepthData
|
|
|
|
try:
|
|
sensor = SensorData.objects.select_related("location").get(
|
|
uuid_sensor=sensor_uuid
|
|
)
|
|
except SensorData.DoesNotExist:
|
|
return None
|
|
|
|
parts: list[str] = []
|
|
|
|
# شناسه سنسور
|
|
parts.append(f"سنسور: {sensor.uuid_sensor}")
|
|
|
|
# موقعیت مزرعه
|
|
loc = sensor.location
|
|
parts.append(
|
|
f"موقعیت مزرعه: عرض {loc.latitude}، طول {loc.longitude}"
|
|
)
|
|
|
|
# خوانشهای سنسور (schema-agnostic)
|
|
sensor_fields = _model_to_data_fields(
|
|
sensor, exclude={"uuid_sensor", "location_id", "location"}
|
|
)
|
|
if sensor_fields:
|
|
sensor_lines = [f" {k}: {v}" for k, v in sorted(sensor_fields.items())]
|
|
parts.append("خوانشهای سنسور:\n" + "\n".join(sensor_lines))
|
|
|
|
# دادههای خاک به تفکیک عمق
|
|
depths = (
|
|
SoilDepthData.objects.filter(soil_location=loc)
|
|
.order_by("depth_label")
|
|
.all()
|
|
)
|
|
if depths:
|
|
depth_parts = []
|
|
for d in depths:
|
|
d_data = _model_to_data_fields(
|
|
d, exclude={"soil_location", "soil_location_id"}
|
|
)
|
|
if d_data:
|
|
lines = [f" {k}: {v}" for k, v in sorted(d_data.items())]
|
|
depth_parts.append(f" عمق {d.depth_label}:\n" + "\n".join(lines))
|
|
if depth_parts:
|
|
parts.append("دادههای خاک:\n" + "\n".join(depth_parts))
|
|
|
|
return "\n\n".join(parts) if parts else None
|
|
|
|
|
|
def get_all_sensor_uuids() -> list[str]:
|
|
"""لیست همه uuid_sensor های موجود."""
|
|
from sensor_data.models import SensorData
|
|
|
|
return [
|
|
str(u) for u in
|
|
SensorData.objects.values_list("uuid_sensor", flat=True).distinct()
|
|
]
|
|
|
|
|
|
def load_user_sources() -> list[tuple[str, str]]:
|
|
"""
|
|
بارگذاری منابع دیتای کاربران از DB.
|
|
Returns: [(source_id, content), ...]
|
|
source_id = user:{sensor_uuid}
|
|
"""
|
|
uuids = get_all_sensor_uuids()
|
|
sources: list[tuple[str, str]] = []
|
|
for uid in uuids:
|
|
text = build_user_soil_text(str(uid))
|
|
if text and text.strip():
|
|
sources.append((f"user:{uid}", text))
|
|
return sources
|