add fan page notify, add twitch route, todo: twitch webhook reg

Dockerfile

@@ -1,10 +1,10 @@
 FROM node:8
-LABEL maintainer="Jay <jie.chen@tw.viewsonic.com>"
+LABEL maintainer="Jay <admin@trj.tw>"
 ENV NODE_PORT 5111
 RUN mkdir -p /data
 WORKDIR /data
 COPY . .
-RUN rm .env
+RUN rm -f .env
 RUN npm install
 EXPOSE ${NODE_PORT}
-CMD ["npm", "start"]
+CMD ["npm", "run", "dbrun"]

background.js

@@ -1,10 +1,56 @@
-const CronJob = require('cron')
+const cron = require('cron')
+const DB = require('./libs/database')
 const fbParser = require('./libs/facebook-pageparse')
+const {
+  pushMessage
+} = require('./libs/line-message')
 
-new CronJob({ // eslint-disable-line
+new cron.CronJob({ // eslint-disable-line
-  cronTime: '00 */10 * * * *',
+  cronTime: '00 */2 * * * *',
-  onTick: () => {
+  onTick: async () => {
-    
+    console.log('run cron')
+    let db = await DB.connect()
+    let text = `select "pageid", "groupid", "lastpost", "notify", "notify_tmpl" from "public"."page_group_rt"`
+    let res = await db.query({
+      text
+    })
+    console.log('rows length :::: ', res.rowCount)
+    if (res.rows.legnth === 0) return
+    await new Promise(resolve => {
+      let count = res.rowCount
+      res.rows.forEach(t => {
+        fbParser.getLastPost(t.pageid)
+          .then((data) => {
+            let n = Math.floor(Date.now() / 1000)
+            if (t.lastpost === data.id || data.time < (n - 10 * 60)) {
+              if (!--count) resolve(null)
+              return
+            }
+            t.lastpost = data.id
+            let msg = t.notify_tmpl || ''
+            if (typeof msg !== 'string' || msg.trim().length === 0) {
+              msg = `${data.txt}\n${data.link}`
+            } else {
+              msg = msg.replace(/{link}/, data.link).replace(/{txt}/, data.txt).replace(/\\n/, '\n')
+            }
+            if (t.notify) {
+              pushMessage(t.groupid, msg).then(() => {}).catch(() => {})
+            }
+            let text = `update "public"."page_group_rt" set "lastpost" = $1, "mtime" = now() where "pageid" = $2 and "groupid" = $3`
+            let values = [data.id, t.pageid, t.groupid]
+            db.query({
+              text,
+              values
+            }).then(() => {}).catch(() => {})
+            if (!--count) resolve(null)
+          })
+          .catch(() => {
+            if (!--count) resolve(null)
+          })
+      })
+    })
+
+    db.release()
   },
   start: true,
   timeZone: 'Asia/Taipei'

bin/dbVersion.json

@@ -0,0 +1,6 @@
+{
+  "versions":[
+    {"file": "main.sql", "version": 1}
+  ],
+  "test": []
+}

bin/dbtool.js

@@ -0,0 +1,80 @@
+/* eslint-disable no-unused-expressions */
+const pg = require('pg')
+const config = require('../config')
+const path = require('path')
+const fs = require('fs')
+
+const dbVersion = require('./dbVersion.json')
+const versions = dbVersion.versions
+const schemaPath = path.resolve(__dirname, '../schema')
+
+if (!Array.isArray(versions) || versions.length === 0) {
+  throw new Error('Schema version empty')
+}
+
+const client = new pg.Client({
+  host: config.database.host,
+  port: config.database.port,
+  user: config.database.user,
+  password: config.database.pass,
+  database: config.database.dbname
+})
+
+// auto start function
+
+!(async function () {
+  let flag = false
+  await client.connect()
+
+  await client.query(`select now()`)
+  console.log('Database Connected')
+
+  if (process.env['NODE_ENV'] === 'test') {
+    if ('test' in dbVersion && Array.isArray(dbVersion.test)) {
+      for (let i in dbVersion.test) {
+        let qstr = fs.readFileSync(path.resolve(schemaPath, dbVersion.test[i].file)).toString()
+        await client.query(qstr)
+      }
+    }
+  }
+
+  let version = -1
+  let checkTable = await client.query(`select exists(select 1 from "information_schema"."tables" where "table_schema" = $1 and "table_name" = $2) as exists`, ['public', 'version_ctrl'])
+  if (checkTable.rows.length > 0 && checkTable.rows[0].exists === true) {
+    let checkVersion = await client.query(`select max(version) as version from "public"."version_ctrl"`)
+    if (checkVersion.rows.length > 0 && 'version' in checkVersion.rows[0] && isFinite(checkVersion.rows[0].version)) {
+      version = checkVersion.rows[0].version
+    }
+  }
+
+  let runVer = versions.filter(t => t.version > version).sort((a, b) => a.version - b.version)
+
+  if (runVer.length === 0) return
+
+  await client.query('begin')
+  try {
+    // write table query
+    for (let i in runVer) {
+      let qstr = fs.readFileSync(path.resolve(schemaPath, runVer[i].file)).toString()
+      await client.query(qstr)
+      let query = `insert into "public"."version_ctrl" ("version", "ctime", "querystr") values ($1, now(), $2)`
+      let param = [runVer[i].version, qstr]
+      await client.query(query, param)
+    }
+
+    await client.query('commit')
+  } catch (err) {
+    flag = true
+    console.log(err)
+    await client.query('rollback')
+  }
+
+  await client.end()
+  if (flag) throw new Error('Not Finish')
+})().then(() => {
+  console.log('Finish')
+  process.exit(0)
+}).catch(err => {
+  console.error(err)
+  process.exit(1)
+})

config/index.js

@@ -3,5 +3,16 @@ module.exports = {
   line: {
     secret: process.env.LINE_SECRET || '',
     access: process.env.LINE_ACCESS || ''
+  },
+  twitch: {
+    clientid: process.env.TWITCH_CLIENT_ID || '',
+    subsecret: process.env.TWITCH_SUB_SECRET || ''
+  },
+  database: {
+    host: process.env.DB_HOST || 'localhost',
+    port: process.env.DB_PORT || 5432,
+    user: process.env.DB_USER || 'postgres',
+    pass: process.env.DB_PASS || '',
+    dbname: process.env.DB_NAME || 'mtfosbot'
   }
 }

index.js

@@ -5,3 +5,4 @@ try {
   require('dotenv').config()
 } catch (err) {}
 require('./app')
+require('./background')

libs/database.js

@@ -0,0 +1,13 @@
+const pg = require('pg')
+const config = require('../config')
+
+const pool = new pg.Pool({
+  user: config.database.user,
+  password: config.database.pass || null,
+  host: config.database.host,
+  port: config.database.port,
+  max: 100,
+  database: config.database.dbname
+})
+
+module.exports = pool

libs/facebook-pageparse/index.js

@@ -1,6 +1,18 @@
 const request = require('request')
 const cheerio = require('cheerio')
 
+/**
+ * @typedef lastPost
+ * @prop {string} txt post body
+ * @prop {string} id post id
+ * @prop {string} link post link
+ * @prop {string} time timestamp
+ */
+/**
+ * get facebook fan page last post
+ * @param {string} pageid facebook fan page id
+ * @return {Promise<lastPost>}
+ */
 const getLastPost = async (pageid = '') => {
   if (typeof pageid !== 'string' || pageid.trim().length === 0) return null
   pageid = pageid.trim()
@@ -38,10 +50,23 @@ const getLastPost = async (pageid = '') => {
     let timeEl = t('abbr')
     let time = timeEl.attr('data-utime')
     let link = timeEl.parent().attr('href')
-    let p = t('div.userContent div.text_exposed_root')
+    let p = t('div.userContent')
-    let txt = p.text()
+    let txt = p.first().text()
-    let id = p.attr('id')
+    let id = p.first().attr('id')
-    if (!time || !link || !txt || !id) return
+    if (!id) {
+      if (/[\?|&]id\=(\d+)/.test(link)) { // eslint-disable-line
+        let m = link.match(/[\?|&]story_fbid\=(\d+)/) // eslint-disable-line
+        if (m !== null && m.length > 1) {
+          id = m[1]
+        }
+      } else if (/\/posts\/(\d+)/.test(link)) {
+        let m = link.match(/\/posts\/(\d+)/)
+        if (m !== null && m.length > 1) {
+          id = m[1]
+        }
+      }
+    }
+    if (!time || !link || !txt || !id) return null
     let tmp = {
       txt,
       id,
@@ -49,9 +74,11 @@ const getLastPost = async (pageid = '') => {
       time
     }
     posts.push(tmp)
+    el = null
+    t = null
   })
-
+  $ = null
-  if (posts.length === 0) return
+  if (posts.length === 0) return null
   posts.sort((a, b) => {
     return b.time - a.time
   })

libs/line-message/index.js

@@ -1,5 +1,6 @@
 const axios = require('axios')
 const config = require('../../config')
+const DB = require('../database')
 
 const client = axios.create({
   baseURL: 'https://api.line.me/v2/bot',
@@ -39,22 +40,24 @@ const textMessage = async (evt) => {
   if (typeof text !== 'string') return
   text = text.trim()
   if (text.length === 0) return
+  let db = await DB.connect()
 
-  let opts = {
+  // let opts = {
-    method: 'post',
+  //   method: 'post',
-    url: replyURL,
+  //   url: replyURL,
-    data: {
+  //   data: {
-      replyToken,
+  //     replyToken,
-      messages: [
+  //     messages: [
-        {
+  //       {
-          type: 'text',
+  //         type: 'text',
-          text: 'test message'
+  //         text: 'test message'
-        }
+  //       }
-      ]
+  //     ]
-    }
+  //   }
-  }
+  // }
 
-  await client(opts)
+  // await client(opts)
+  db.release()
 }
 
 module.exports = {

libs/middleware/index.js

@@ -1,4 +1,5 @@
 const config = require('../../config')
+const rawBody = require('raw-body')
 const crypto = require('crypto')
 
 const verifyLine = async (c, n) => {
@@ -10,6 +11,25 @@ const verifyLine = async (c, n) => {
   return n()
 }
 
-module.exports = {
+const getRaw = async (c, n) => {
-  verifyLine
+  let raw = await rawBody(c.req, {
+    length: c.request.length,
+    limit: '5mb',
+    encoding: c.request.charset
+  })
+  c.request.raw = raw
+  let txt = raw instanceof Buffer ? raw.toString() : raw
+  if (c.request.type === 'application/json') {
+    try {
+      c.request.body = JSON.parse(txt)
+    } catch (err) {
+      c.request.body = txt
+    }
+  }
+  return n()
+}
+
+module.exports = {
+  verifyLine,
+  getRaw
 }

package.json

@@ -5,6 +5,8 @@
   "main": "index.js",
   "scripts": {
     "start": "node index.js 2>&1 | tee runtime.txt",
+    "dbtool": "node bin/dbtool",
+    "dbrun": "npm run dbtool && npm start",
     "test": "echo \"Error: no test specified\" && exit 1"
   },
   "keywords": [],
@@ -19,6 +21,8 @@
     "koa-body": "^4.0.3",
     "koa-logger": "^3.2.0",
     "koa-router": "^7.4.0",
+    "pg": "^7.4.3",
+    "raw-body": "^2.3.3",
     "request": "^2.87.0"
   },
   "devDependencies": {

route/index.js

@@ -2,6 +2,6 @@ const Router = require('koa-router')
 const r = new Router()
 
 r.use('/line', require('./line').routes())
-r.use('/fb', require('./facebook').routes())
+r.use('/twitch', require('./twitch').routes())
 
 module.exports = r

route/line/index.js

@@ -1,32 +1,14 @@
 const Router = require('koa-router')
 const r = new Router()
 // const koaBody = require('koa-body')
-const rawBody = require('raw-body')
 const {
-  verifyLine
+  verifyLine,
+  getRaw
 } = require('../../libs/middleware')
 const {
   textMessage
 } = require('../../libs/line-message')
 
-const getRaw = async (c, n) => {
-  let raw = await rawBody(c.req, {
-    length: c.request.length,
-    limit: '5mb',
-    encoding: c.request.charset
-  })
-  c.request.raw = raw
-  let txt = raw instanceof Buffer ? raw.toString() : raw
-  if (c.request.type === 'application/json') {
-    try {
-      c.request.body = JSON.parse(txt)
-    } catch (err) {
-      c.request.body = txt
-    }
-  }
-  return n()
-}
-
 r.post('/', getRaw, verifyLine, async (c, n) => {
   console.log(JSON.stringify(c.request.body, null, 2))
   if (!('events' in c.request.body)) return c.throw(400, 'data struct error')

route/twitch/index.js

@@ -0,0 +1,31 @@
+const Router = require('koa-router')
+const r = new Router()
+const {
+  getRaw
+} = require('../../libs/middleware')
+const config = require('../../config')
+
+r.get('/', async (c, n) => {
+  let mode = c.query['hub.mode']
+  let token = c.query['hub.secret']
+  let challenge = c.query['hub.challenge']
+  console.log(mode, token, challenge)
+  console.log(c.headers)
+  if (mode) {
+    if (mode === 'subscribe') {
+      c.status = 200
+      c.body = challenge
+    } else {
+      c.status = 403
+      c.body = ''
+    }
+  }
+})
+
+r.post('/', getRaw, async (c, n) => {
+  console.log(JSON.stringify(c.request.body, null, 2))
+  c.body = 'success'
+  c.status = 200
+})
+
+module.exports = r

schema/main.sql

@@ -0,0 +1,102 @@
+--
+-- PostgreSQL database dump
+--
+
+-- Dumped from database version 10.2 (Debian 10.2-1.pgdg90+1)
+-- Dumped by pg_dump version 10.2 (Debian 10.2-1.pgdg90+1)
+
+SET statement_timeout = 0;
+SET lock_timeout = 0;
+SET idle_in_transaction_session_timeout = 0;
+SET client_encoding = 'UTF8';
+SET standard_conforming_strings = on;
+SET check_function_bodies = false;
+SET client_min_messages = warning;
+SET row_security = off;
+
+SET search_path = public, pg_catalog;
+
+ALTER TABLE IF EXISTS ONLY public.page_group_rt DROP CONSTRAINT IF EXISTS page_group_rt_pageid_groupid_pk;
+DROP TABLE IF EXISTS public.version_ctrl;
+DROP TABLE IF EXISTS public.page_group_rt;
+DROP EXTENSION IF EXISTS plpgsql;
+DROP SCHEMA IF EXISTS public;
+--
+-- Name: public; Type: SCHEMA; Schema: -; Owner: -
+--
+
+CREATE SCHEMA public;
+
+
+--
+-- Name: SCHEMA public; Type: COMMENT; Schema: -; Owner: -
+--
+
+COMMENT ON SCHEMA public IS 'standard public schema';
+
+
+--
+-- Name: plpgsql; Type: EXTENSION; Schema: -; Owner: -
+--
+
+CREATE EXTENSION IF NOT EXISTS plpgsql WITH SCHEMA pg_catalog;
+
+
+--
+-- Name: EXTENSION plpgsql; Type: COMMENT; Schema: -; Owner: -
+--
+
+COMMENT ON EXTENSION plpgsql IS 'PL/pgSQL procedural language';
+
+
+SET search_path = public, pg_catalog;
+
+SET default_tablespace = '';
+
+SET default_with_oids = false;
+
+--
+-- Name: page_group_rt; Type: TABLE; Schema: public; Owner: -
+--
+
+CREATE TABLE page_group_rt (
+    pageid character varying(200) NOT NULL,
+    groupid character varying(200) NOT NULL,
+    notify boolean DEFAULT false NOT NULL,
+    lastpost character varying(200) DEFAULT ''::character varying NOT NULL,
+    ctime timestamp with time zone DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    mtime timestamp with time zone DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    notify_tmpl character varying(500) DEFAULT ''::character varying NOT NULL
+);
+
+
+--
+-- Name: version_ctrl; Type: TABLE; Schema: public; Owner: -
+--
+
+CREATE TABLE version_ctrl (
+    version integer NOT NULL,
+    ctime timestamp with time zone DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    querystr character varying(5000) DEFAULT ''::character varying NOT NULL
+);
+
+
+--
+-- Name: page_group_rt page_group_rt_pageid_groupid_pk; Type: CONSTRAINT; Schema: public; Owner: -
+--
+
+ALTER TABLE ONLY page_group_rt
+    ADD CONSTRAINT page_group_rt_pageid_groupid_pk PRIMARY KEY (pageid, groupid);
+
+
+--
+-- Name: SCHEMA public; Type: ACL; Schema: -; Owner: -
+--
+
+GRANT ALL ON SCHEMA public TO PUBLIC;
+
+
+--
+-- PostgreSQL database dump complete
+--
+