Merge pull request #554 from CodeForPhilly/538-run_dates

API for last update times

Author: c-simpson

src/server/alembic/versions/d80cb6df0fa2_rmv_shifts_uniqueness_constraint.py

@@ -0,0 +1,32 @@
+"""rmv shifts uniqueness constraint
+
+Revision ID: d80cb6df0fa2
+Revises: 90f471ac445c
+Create Date: 2023-03-18 16:22:23.282568
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'd80cb6df0fa2'
+down_revision = '90f471ac445c'
+branch_labels = None
+depends_on = None
+
+# It's probably more likely that a duplicate row is actually a real shift with a bad (dupe)
+# like Saturday, Saturday instead of Saturday, Sunday 
+# We really care about last shift so this is not critical
+
+def upgrade():
+    op.drop_constraint( "uq_shift", "volgisticsshifts") 
+
+def downgrade():
+    # op.create_unique_constraint( "uq_shift", "volgisticsshifts",  ["volg_id", "assignment", "from_date", "hours"] ) 
+    # This will fail if you have any dupes
+    # running 
+    #    ALTER TABLE "public"."volgisticsshifts" ADD CONSTRAINT "uq_shift" UNIQUE( "volg_id", "assignment", "from_date", "hours" );
+    # will fail and tell you of any dupes so you can fix
+
+    pass

src/server/api/API_ingest/ingest_sources_from_api.py

@@ -1,5 +1,7 @@
 from api.API_ingest import shelterluv_people, salesforce_contacts, sl_animal_events
 import structlog
+
+from pipeline.log_db import log_shelterluv_update
 logger = structlog.get_logger()
 
 def start():
@@ -16,6 +18,7 @@ def start():
     logger.debug("   Fetching Shelterluv events")
     sle_count = sl_animal_events.store_all_animals_and_events()
     logger.debug("   Finished fetching Shelterluv events - %d records" , sle_count)
+    log_shelterluv_update()
 
     logger.debug("Finished fetching raw data from different API sources")
 

src/server/api/API_ingest/volgistics_db.py

@@ -0,0 +1,52 @@
+from sqlalchemy import Table, MetaData
+from sqlalchemy.orm import sessionmaker
+
+from config import engine
+
+import structlog
+logger = structlog.get_logger()
+
+def insert_volgistics_people(row_list):
+
+    row_count = 0
+    try:
+        Session = sessionmaker(engine)
+        session = Session()
+        metadata = MetaData()
+        volg_table = Table("volgistics", metadata, autoload=True, autoload_with=engine)
+
+        result = session.execute("TRUNCATE table volgistics;")
+        ret = session.execute(volg_table.insert(row_list))
+
+        row_count = ret.rowcount
+
+        session.commit()  # Commit all inserted rows
+        session.close()
+    except Exception as e:
+        row_count = 0
+        logger.error("Exception inserting volgistics people")
+        logger.exception(e)
+    return row_count
+
+
+def insert_volgistics_shifts(row_list):
+
+    row_count = 0
+    try:
+        Session = sessionmaker(engine)
+        session = Session()
+        metadata = MetaData()
+        volg_table = Table("volgisticsshifts", metadata, autoload=True, autoload_with=engine)
+
+        result = session.execute("TRUNCATE table volgisticsshifts;")
+        ret = session.execute(volg_table.insert(row_list))
+
+        row_count = ret.rowcount
+
+        session.commit()  # Commit all inserted rows
+        session.close()
+    except Exception as e:
+        row_count = 0
+        logger.error("Exception inserting volgistics shifts")
+        logger.exception(e.pgerror)
+    return row_count

src/server/api/admin_api.py

@@ -231,6 +231,26 @@ def start_job():
         return job_id
 
 
+@admin_api.route("/api/get_last_runs", methods=["GET"])
+#@jwt_ops.admin_required
+def get_run_logs():
+    """ Get the timestamps of the last update runs"""
+
+    with engine.connect() as connection:
+        q = text("""select keycol,valcol from kv_unique where keycol like '%_update'; """)
+        result = connection.execute(q)
+
+        if result.rowcount > 0:
+           rows = result.fetchall()
+
+        row_list = []
+
+        for row in rows:
+            row_dict = row._mapping 
+            row_list.append({row_dict['keycol'] : row_dict['valcol']})
+
+        return jsonify(row_list)
+
 
 def insert_rfm_scores(score_list):
     """Take a list of (matching_id, score) and insert into the

src/server/api/file_uploader.py

@@ -1,11 +1,10 @@
-import pandas as pd
 from config import engine
 from donations_importer import validate_import_sfd
 from flask import current_app
 from models import ManualMatches, SalesForceContacts, ShelterluvPeople, Volgistics
-from shifts_importer import validate_import_vs
+from pipeline.log_db import log_volgistics_update
+from volgistics_importer import open_volgistics, validate_import_vs, volgistics_people_import
 from werkzeug.utils import secure_filename
-
 import structlog
 logger = structlog.get_logger()
 
@@ -26,39 +25,17 @@ def determine_upload_type(file, file_extension, conn):
     # automatically pulling from vendor APIs directly, in which case we'd know
     # what kind of data we had.
     if file_extension == "csv":
-        logger.debug("File extension is CSV")
-        df = pd.read_csv(file, dtype="string")
-
-        if {"salesforcecontacts", "volgistics", "shelterluvpeople"}.issubset(df.columns):
-            logger.debug("File appears to be salesforcecontacts, volgistics, or shelterluvpeople (manual)")
-            ManualMatches.insert_from_df(df, conn)
-            return
-        elif {"Animal_ids", "Internal-ID"}.issubset(df.columns):
-            logger.debug("File appears to be shelterluvpeople")
-            ShelterluvPeople.insert_from_df(df, conn)
-            return
+        logger.warn("%s: We no longer support CSV files", file.filename)
+        return
 
     if file_extension == "xlsx":
-        excel_file = pd.ExcelFile(file)
-        if {"Master", "Service"}.issubset(excel_file.sheet_names):
-            logger.debug("File appears to be Volgistics")
-            # Volgistics
-            validate_import_vs(file, conn)
-            Volgistics.insert_from_file(excel_file, conn)
-            return
-
-        df = pd.read_excel(excel_file)
-        if "Contact ID 18" in df.columns:
-            # Salesforce something-or-other
-            if "Amount" in df.columns:
-                # Salesforce donations
-                logger.debug("File appears to be Salesforce donations")
-                validate_import_sfd(file, conn)
-                return
-            else:
-                # Salesforce contacts
-                logger.debug("File appears to be Salesforce contacts")
-                SalesForceContacts.insert_from_file_df(df, conn)
-                return
+        # Assume it's Volgistics
+        workbook = open_volgistics(file)
+        if workbook:
+            validate_import_vs(workbook)
+            volgistics_people_import(workbook)
+            workbook.close()
+            log_volgistics_update()
+        return
 
     logger.error("Don't know how to process file: %s",  file.filename)

src/server/api/internal_api.py

@@ -5,7 +5,7 @@
 
 from api.API_ingest import ingest_sources_from_api, salesforce_contacts
 from api.api import internal_api
-from rfm_funcs.create_scores import create_scores
+# from rfm_funcs.create_scores import create_scores
 from api.API_ingest import updated_data
 
 logger = structlog.get_logger()
@@ -37,12 +37,12 @@ def ingest_raw_data():
     return jsonify({'outcome': 'OK'}), 200
 
 
-@internal_api.route("/api/internal/create_scores", methods=["GET"])
-def hit_create_scores():
-    logger.info("Hitting create_scores() ")
-    tuple_count = create_scores()
-    logger.info("create_scores()  processed %s scores",  str(tuple_count) )
-    return jsonify(200)
+# @internal_api.route("/api/internal/create_scores", methods=["GET"])
+# def hit_create_scores():
+#     logger.info("Hitting create_scores() ")
+#     tuple_count = create_scores()
+#     logger.info("create_scores()  processed %s scores",  str(tuple_count) )
+#     return jsonify(200)
 
 
 @internal_api.route("/api/internal/get_updated_data", methods=["GET"])

src/server/models.py

@@ -2,7 +2,6 @@
 import re
 from itertools import combinations
 
-import pandas as pd
 import sqlalchemy as sa
 from sqlalchemy import (
     Boolean,
@@ -296,35 +295,35 @@ class Volgistics(Base):
     json = Column(JSONB)
     created_date = Column(DateTime, default=datetime.datetime.utcnow)
 
-    @classmethod
-    def insert_from_file(cls, xl_file, conn):
-        df = pd.read_excel(xl_file, sheet_name="Master")
-
-        column_translation = get_source_column_translation(cls)
-        df = df[column_translation.keys()]
-        df = df.rename(columns=column_translation)
-
-        df["home"] = df["home"].apply(normalize_phone_number)
-        df["work"] = df["work"].apply(normalize_phone_number)
-        df["cell"] = df["cell"].apply(normalize_phone_number)
-
-        dedup_on = [col for col in cls.__table__.columns if col.name in df.columns]
-        df["created_date"] = datetime.datetime.utcnow()
-        df.to_sql(
-            cls.__tablename__,
-            conn,
-            if_exists="append",
-            index=False,
-        )
-        conn.execute(
-            dedup_consecutive(
-                cls.__table__,
-                unique_id=cls._id,
-                id=cls.number,
-                order_by=cls.created_date,
-                dedup_on=tuple_(*dedup_on),
-            )
-        )
+    # @classmethod
+    # def insert_from_file(cls, xl_file, conn):
+    #     df = pd.read_excel(xl_file, sheet_name="Master")
+
+    #     column_translation = get_source_column_translation(cls)
+    #     df = df[column_translation.keys()]
+    #     df = df.rename(columns=column_translation)
+
+    #     df["home"] = df["home"].apply(normalize_phone_number)
+    #     df["work"] = df["work"].apply(normalize_phone_number)
+    #     df["cell"] = df["cell"].apply(normalize_phone_number)
+
+    #     dedup_on = [col for col in cls.__table__.columns if col.name in df.columns]
+    #     df["created_date"] = datetime.datetime.utcnow()
+    #     df.to_sql(
+    #         cls.__tablename__,
+    #         conn,
+    #         if_exists="append",
+    #         index=False,
+    #     )
+    #     conn.execute(
+    #         dedup_consecutive(
+    #             cls.__table__,
+    #             unique_id=cls._id,
+    #             id=cls.number,
+    #             order_by=cls.created_date,
+    #             dedup_on=tuple_(*dedup_on),
+    #         )
+    #     )
 
     @classmethod
     def insert_into_pdp_contacts(cls):
@@ -356,29 +355,29 @@ class ManualMatches(Base):
     source_type_2 = Column(String, primary_key=True)
     source_id_2 = Column(String, primary_key=True)
 
-    @classmethod
-    def insert_from_df(cls, df, conn):
-        # Our input csv has columns like "salesforcecontacts," "volgistics," and
-        # "shelterluvpeople," where two columns are non-null if there is an
-        # association between those two ids. We massage this table into one that
-        # is easier to join on.
+    # @classmethod
+    # def insert_from_df(cls, df, conn):
+    #     # Our input csv has columns like "salesforcecontacts," "volgistics," and
+    #     # "shelterluvpeople," where two columns are non-null if there is an
+    #     # association between those two ids. We massage this table into one that
+    #     # is easier to join on.
 
-        match_dicts = df.to_dict(orient="records")
-
-        matched_pairs = []
-        for match in match_dicts:
-            non_nulls = {k: v for (k, v) in match.items() if not pd.isna(v)}
-            for ((st1, sid1), (st2, sid2)) in combinations(non_nulls.items(), 2):
-                matched_pairs.append(
-                    {
-                        "source_type_1": st1,
-                        "source_id_1": sid1,
-                        "source_type_2": st2,
-                        "source_id_2": sid2,
-                    }
-                )
-
-        conn.execute(insert(cls).values(matched_pairs).on_conflict_do_nothing())
+    #     match_dicts = df.to_dict(orient="records")
+
+    #     matched_pairs = []
+    #     for match in match_dicts:
+    #         non_nulls = {k: v for (k, v) in match.items() if not pd.isna(v)}
+    #         for ((st1, sid1), (st2, sid2)) in combinations(non_nulls.items(), 2):
+    #             matched_pairs.append(
+    #                 {
+    #                     "source_type_1": st1,
+    #                     "source_id_1": sid1,
+    #                     "source_type_2": st2,
+    #                     "source_id_2": sid2,
+    #                 }
+    #             )
+
+    #     conn.execute(insert(cls).values(matched_pairs).on_conflict_do_nothing())
 
 class SalesforceDonations(Base):
     __tablename__ = "salesforcedonations"