import { mongoConnect , mongoDisconnect } from "./mongoose" ; import _ from "lodash" ; import moment from "moment" ; import { forEachSeries } from "p-iteration" ; import { PROCEDURE as PROCEDURE_DEFINITIONS } from "@democracy-deutschland/bundestag.io-definitions" ; import createClient from "./graphql/client" ; import getProcedureUpdates from "./graphql/queries/getProcedureUpdates" ; import { ProcedureModel , VoteSelection , getCron , setCronStart , setCronSuccess , setCronError , queuePushsOutcome , convertPartyName , IProcedure , PartyVotes , ProcedureDocument , } from "@democracy-deutschland/democracy-common" ; import { ProcedureUpdates , ProcedureUpdatesVariables , ProcedureUpdates_procedureUpdates_procedures , } from "./graphql/queries/__generated__/ProcedureUpdates" ; import { VoteDecision } from "./__generated__/globalTypes" ; export const CRON_NAME = "Procedures" ; const notEmpty = < TValue > ( value ) : value is TValue => { return value !== null && value !== undefined ; } ; export const nullToUndefined = < TValue > ( value ) => { return value === null ? undefined : value ; } ; const importProcedures = async ( bIoProcedure , { push = false } ) => { const importProcedure = { ... bIoProcedure , procedureId : nullToUndefined ( bIoProcedure . procedureId ) , period : nullToUndefined ( bIoProcedure . period ) , type : nullToUndefined ( bIoProcedure . type ) , currentStatus : nullToUndefined ( bIoProcedure . currentStatus ) , abstract : nullToUndefined ( bIoProcedure . abstract ) , currentStatusHistory : bIoProcedure . currentStatusHistory ? bIoProcedure . currentStatusHistory . filter ( notEmpty ) : undefined , tags : bIoProcedure . tags ? bIoProcedure . tags . filter ( notEmpty ) : undefined , subjectGroups : bIoProcedure . subjectGroups ? bIoProcedure . subjectGroups . filter ( notEmpty ) : undefined , importantDocuments : bIoProcedure . importantDocuments ?. reduce < ProcedureDocument [ ] > ( ( prev , doc ) => { if ( doc ) { return [ ... prev , doc ] as ProcedureDocument [ ] ; } return prev ; } , [ ] ) , } ; if ( bIoProcedure && bIoProcedure . history ) { const [ lastHistory ] = bIoProcedure . history . slice ( - 1 ) ; if ( lastHistory ) { importProcedure . lastUpdateDate = lastHistory . date ; } if ( bIoProcedure . history [ 0 ] ) { importProcedure . submissionDate = bIoProcedure . history [ 0 ] . date ; } } let voteResults ; if ( bIoProcedure . customData && bIoProcedure . customData . voteResults && ( bIoProcedure . customData . voteResults . yes || bIoProcedure . customData . voteResults . abstination || bIoProcedure . customData . voteResults . no ) ) { voteResults = { yes : bIoProcedure . customData . voteResults . yes || 0 , abstination : bIoProcedure . customData . voteResults . abstination || 0 , no : bIoProcedure . customData . voteResults . no || 0 , notVoted : nullToUndefined ( bIoProcedure . customData . voteResults . notVoted ) , decisionText : nullToUndefined ( bIoProcedure . customData . voteResults . decisionText ) , namedVote : nullToUndefined ( bIoProcedure . namedVote ) , partyVotes : [ ] , } ; if ( bIoProcedure . customData . voteResults . partyVotes ) { voteResults . partyVotes = bIoProcedure . customData . voteResults . partyVotes . reduce < PartyVotes [ ] > ( ( pre , partyVote ) => { if ( partyVote ) { let mainDecision ; const { main , party , ... rest } = partyVote ; switch ( main ) { case VoteDecision . YES : mainDecision = VoteSelection . Yes ; break ; case VoteDecision . ABSTINATION : mainDecision = VoteSelection . Abstination ; break ; case VoteDecision . NO : mainDecision = VoteSelection . No ; break ; default : mainDecision = VoteSelection . Notvoted ; } let deviants ; if ( rest . deviants && rest . deviants . yes !== null && rest . deviants . abstination !== null && rest . deviants . no !== null ) { deviants = { yes : rest . deviants . yes , abstination : rest . deviants . abstination , no : rest . deviants . no , notVoted : rest . deviants . notVoted , } ; } if ( ! deviants ) { return pre ; } const result = { ... rest , _id : false , party : convertPartyName ( party ) , main : mainDecision , deviants , } ; return [ ... pre , result ] ; } return pre ; } , [ ] ) ; if ( bIoProcedure . customData . voteResults . votingDocument === "recommendedDecision" && bIoProcedure . customData . voteResults . votingRecommendation === false ) { voteResults = { ... voteResults , yes : voteResults . no , no : voteResults . yes , partyVotes : voteResults . partyVotes . map ( ( { main , deviants , ... rest } ) => { let mainDecision = main ; if ( main !== "ABSTINATION" ) { mainDecision = main === VoteSelection . Yes ? VoteSelection . No : VoteSelection . Yes ; } return { ... rest , main : mainDecision , deviants : { ... deviants , yes : deviants . no , no : deviants . yes , } , } ; } ) , } ; } } } importProcedure . voteResults = voteResults ; if ( bIoProcedure . sessions ) { const lastSession = bIoProcedure . sessions . pop ( ) ; if ( lastSession && lastSession . session ?. top ?. topic ?. isVote ) { importProcedure . voteWeek = lastSession . thisWeek ; importProcedure . voteYear = lastSession . thisYear ; importProcedure . sessionTOPHeading = nullToUndefined ( lastSession . session . top . heading ) ; } } if ( bIoProcedure . voteDate ) { importProcedure . voteWeek = parseInt ( moment ( bIoProcedure . voteDate ) . format ( "W" ) ) ; importProcedure . voteYear = moment ( bIoProcedure . voteDate ) . year ( ) ; } const oldProcedure = await ProcedureModel . findOne ( { procedureId : nullToUndefined ( bIoProcedure . procedureId ) , } ) ; return ProcedureModel . findOneAndUpdate ( { procedureId : importProcedure . procedureId } , _ ( importProcedure ) . omitBy ( ( x ) => _ . isUndefined ( x ) ) . value ( ) , { upsert : true , new : true , } ) . then ( async ( ) => { if ( push ) { if ( importProcedure . voteResults && ( importProcedure . voteResults . yes !== null || importProcedure . voteResults . no !== null || importProcedure . voteResults . abstination !== null || importProcedure . voteResults . notVoted !== null ) ) { if ( importProcedure . procedureId && ( ! oldProcedure || ! oldProcedure . voteResults ) ) { await queuePushsOutcome ( importProcedure . procedureId ) ; } else if ( importProcedure . procedureId && importProcedure . voteResults && oldProcedure && ( importProcedure . voteResults . yes !== oldProcedure . voteResults . yes || importProcedure . voteResults . no !== oldProcedure . voteResults . no || importProcedure . voteResults . abstination !== oldProcedure . voteResults . abstination || ( importProcedure . voteResults . notVoted && importProcedure . voteResults . notVoted !== oldProcedure . voteResults . notVoted ) ) ) { await queuePushsOutcome ( importProcedure . procedureId ) ; } } } } ) ; } ; const start = async ( ) => { const startDate = new Date ( ) ; const cron = await getCron ( { name : CRON_NAME } ) ; await setCronStart ( { name : CRON_NAME , startDate } ) ; let since = new Date ( "1900" ) ; if ( cron . lastSuccessStartDate ) { since = new Date ( cron . lastSuccessStartDate ) ; } let counter = 0 ; try { const client = createClient ( ) ; const limit = process . env . LIMIT ? parseInt ( process . env . LIMIT ) : 25 ; let offset = 0 ; let done = false ; while ( ! done ) { const variables = { since , limit , offset , periods : [ 18 , 19 , 20 ] , types : [ PROCEDURE_DEFINITIONS . TYPE . GESETZGEBUNG , PROCEDURE_DEFINITIONS . TYPE . ANTRAG , ] , } ; const { errors , data : { procedureUpdates } , } = await client . query < ProcedureUpdates , ProcedureUpdatesVariables > ( { query : getProcedureUpdates , variables , } ) ; console . log ( "use variables:" , { ... variables , progress : { ... ( procedureUpdates ? { afterCount : procedureUpdates . afterCount , beforeCount : procedureUpdates . beforeCount , changedCount : procedureUpdates . changedCount , newCount : procedureUpdates . newCount , } : { } ) , } , } ) ; if ( procedureUpdates ) { const { procedures } = procedureUpdates ; if ( procedures ) { counter += procedures . length ; await forEachSeries ( procedures , async ( data ) => { if ( data && ( data . type === PROCEDURE_DEFINITIONS . TYPE . GESETZGEBUNG || data . type === PROCEDURE_DEFINITIONS . TYPE . ANTRAG ) ) { await importProcedures ( data , { push : true } ) ; } } ) ; if ( procedures . length < limit ) { done = true ; } offset += limit ; } } else { await setCronError ( { name : CRON_NAME , error : JSON . stringify ( errors ) } ) ; throw errors ; } } await setCronSuccess ( { name : CRON_NAME , successStartDate : startDate } ) ; console . log ( `synced items: ${ counter }` ) ; } catch ( error ) { console . error ( error ) ; await setCronError ( { name : CRON_NAME , error : JSON . stringify ( error ) } ) ; throw error ; } } ; ( async ( ) => { console . info ( "START" ) ; console . info ( "process.env" , process . env . BUNDESTAGIO_SERVER_URL , process . env . DB_URL ) ; if ( ! process . env . BUNDESTAGIO_SERVER_URL ) { throw new Error ( "you have to set environment variable: BUNDESTAGIO_SERVER_URL & DB_URL" ) ; } await mongoConnect ( ) ; console . log ( "procedures" , await ProcedureModel . countDocuments ( { } ) ) ; await start ( ) ; await mongoDisconnect ( ) ; } ) ( ) . catch ( async ( e ) => { await mongoDisconnect ( ) ; throw e ; } ) ;