Add experimental "comments daemon".

git-svn-id: http://opensvn.csie.org/pttbbs/trunk@5912 63ad8ddf-47c3-0310-b6dd-a9e9d9715204

3 files changed, 202 insertions, 1 deletions

diff --git a/pttbbs/daemon/commentd/commentd.py b/pttbbs/daemon/commentd/commentd.py
new file mode 100755
index 00000000..30cf13bf
--- /dev/null
+++ b/pttbbs/daemon/commentd/commentd.py
@@ -0,0 +1,129 @@
+#!/usr/bin/env python
+
+import struct
+import collections
+import sys
+import logging
+import os
+
+from gevent import monkey; monkey.patch_all()
+import gevent
+import gevent.server
+import leveldb
+
+# Ref: ../../include/pttstruct.h
+IDLEN = 12
+FNLEN = 28
+COMMENTLEN = 80
+
+
+# Ref: ../../include/daemons.h
+RequestFormatString = 'HH';
+Request = collections.namedtuple('Request', 'cb operation')
+CommentFormatString = "III%ds%ds" % (IDLEN + 1, COMMENTLEN + 1)
+Comment = collections.namedtuple('Comment',
+	'time ipv4 userref type userid msg')
+CommentKeyFormatString = '%ds%ds' % (IDLEN + 1, FNLEN + 1)
+CommentKey = collections.namedtuple('CommentKey', 'board file')
+REQ_ADD = 1
+REQ_QUERY = 2
+_SERVER_ADDR = '127.0.0.1'
+_SERVER_PORT = 5134
+_DB_PATH = './db_comments.db'
+
+def UnpackComment(blob):
+    def strip_if_string(v):
+	return v.strip(chr(0)) if type(v) == str else v
+    data = struct.unpack(CommentFormatString, blob)
+    logging.debug("UnpackComment: %r" % (data,))
+    return Comment._make(map(strip_if_string, data))
+
+def UnpackCommentKey(blob):
+    def strip_if_string(v):
+	return v.strip(chr(0)) if type(v) == str else v
+    data = struct.unpack(CommentKeyFormatString, blob)
+    logging.debug("UnpackCommentKey: %r" % (data,))
+    return CommentKey._make(map(strip_if_string, data))
+
+def PackComment(comment):
+    return struct.pack(CommentFormatString, *comment)
+
+def LoadComment(key):
+    blob = g_db.get(key)
+    if blob is None:
+	return ''
+    else:
+	return UnpackComment(g_db.get(key))
+
+def SaveComment(keypak, comment):
+    logging.debug("SaveComment: %r => %r" % (keypak, comment))
+    key = '%s/%s' % (keypak.board, keypak.file)
+    num = g_db.get(key)
+    num = 1 if (num is None) else (int(num) + 1)
+    g_db.set(key, '%d' % num)
+    key += '#%08d' % (num)
+    g_db.set(key, PackComment(comment))
+    logging.info('Saved comment: %s', key)
+
+def open_database(db_path):
+    global g_db
+
+    class LevelDBWrapper(object):
+	def __init__(self, db):
+	    self.db = db
+
+	def get(self, key):
+	    try:
+		return self.db.Get(key)
+	    except KeyError:
+		return None
+
+	def set(self, key, value):
+	    self.db.Put(key, value)
+
+    g_db = LevelDBWrapper(leveldb.LevelDB(db_path))
+    return g_db
+
+def handle_request(socket, _):
+    fd = socket.makefile('rw')
+    fmt_len = '@i'
+    try:
+        req = fd.read(struct.calcsize(RequestFormatString))
+	req = Request._make(struct.unpack(RequestFormatString, req))
+	logging.debug('Found request: %d' % req.operation)
+        # TODO check req.cb
+	if req.operation == REQ_ADD:
+	    blob = fd.read(struct.calcsize(CommentFormatString))
+	    keyblob = fd.read(struct.calcsize(CommentKeyFormatString))
+	    SaveComment(UnpackCommentKey(keyblob), UnpackComment(blob))
+	elif req.operation == REQ_QUERY:
+	    raise NotImplementedError('REQ_QUERY is not implemented')
+	else:
+	    raise ValueError('Unkown operation %d' % req.operation)
+    except:
+        logging.exception("handle_request")
+    finally:
+        try:
+            fd.close()
+            socket.close()
+        except:
+            pass
+
+def main(myname, argv):
+    level = logging.INFO
+    level = logging.WARNING
+    level = logging.DEBUG
+    logging.basicConfig(level=level, format='%(asctime)-15s %(message)s')
+    if len(argv) not in [0, 1]:
+        print "Usage: %s [db_path]" % myname
+        exit(1)
+    db_path = argv[0] if len(argv) > 0 else _DB_PATH
+    logging.warn("Serving at %s:%s [db:%s][pid:%d]...",
+                 _SERVER_ADDR, _SERVER_PORT, db_path, os.getpid())
+    open_database(db_path)
+    server = gevent.server.StreamServer(
+	    (_SERVER_ADDR, _SERVER_PORT), handle_request)
+    server.serve_forever()
+
+if __name__ == '__main__':
+    main(sys.argv[0], sys.argv[1:])
diff --git a/pttbbs/include/daemons.h b/pttbbs/include/daemons.h
index 545462a2..a2162f29 100644
--- a/pttbbs/include/daemons.h
+++ b/pttbbs/include/daemons.h
@@ -132,6 +132,44 @@ enum BRCSTORED_OPERATIONS {
 };
 
 ///////////////////////////////////////////////////////////////////////
+// Comments Daemon
+
+#ifndef COMMENTD_ADDR
+#define COMMENTD_ADDR   ":5134"
+#endif
+
+#ifndef COMMENTLEN
+#define COMMENTLEN (80)
+#endif
+
+enum {
+    COMMENTD_REQ_ADD = 1,
+    COMMENTD_REQ_QUERY,
+};
+
+typedef struct {
+    time4_t time;
+    time4_t ipv4;
+    uint32_t userref; /* user.ctime */
+    uint32_t type;
+    char userid[IDLEN + 1];
+    char msg[COMMENTLEN + 1];
+} PACKSTRUCT CommentBodyReq;
+
+typedef struct {
+    char board[IDLEN + 1];
+    char file[FNLEN + 1];
+} PACKSTRUCT CommentKeyReq;
+
+typedef struct {
+    short cb;
+    short operation;
+    CommentBodyReq comment;
+    CommentKeyReq key;
+} PACKSTRUCT CommentRequest;
+
+
+///////////////////////////////////////////////////////////////////////
 // online friend relation daemon
 //
 typedef struct {
diff --git a/pttbbs/mbbsd/bbs.c b/pttbbs/mbbsd/bbs.c
index 9c901859..f5aeb609 100644
--- a/pttbbs/mbbsd/bbs.c
+++ b/pttbbs/mbbsd/bbs.c
@@ -1,5 +1,7 @@
 
 #include "bbs.h"
+#include "daemons.h"
+#include <arpa/inet.h>
 
 #ifdef EDITPOST_SMARTMERGE
 
@@ -3112,10 +3114,42 @@ recommend(int ent, fileheader_t * fhdr, const char *direct)
         } else
             return FULLUPDATE;
     }
+    STATINC(STAT_RECOMMEND);
     LOG_IF(LOG_CONF_PUSH, log_filef("log/push", LOG_CREAT,
                                     "%s %d %s %s %s\n", cuser.userid,
                                     (int)now, currboard, fhdr->filename, msg));
-    STATINC(STAT_RECOMMEND);
+#ifdef USE_COMMENTD
+    {
+        int s;
+        const char *errmsg = "錯誤: 資料庫連線異常，無法寫入。請稍候再試。";
+        CommentRequest req = {0};
+        req.cb = sizeof(req);
+        req.operation = COMMENTD_REQ_ADD;
+
+        strlcpy(req.key.board, bp->brdname, sizeof(req.key.board));
+        strlcpy(req.key.file, fhdr->filename, sizeof(req.key.file));
+
+        req.comment.time = now;
+        req.comment.ipv4 = inet_addr(fromhost);
+        req.comment.userref = cuser.firstlogin;
+        req.comment.type = type;
+        strlcpy(req.comment.userid, cuser.userid, sizeof(req.comment.userid));
+        strlcpy(req.comment.msg, msg, sizeof(req.comment.msg));
+
+        s = toconnectex(COMMENTD_ADDR, 10);
+        if (s < 0) {
+            vmsg(errmsg);
+            return FULLUPDATE;
+        }
+        if (towrite(s, &req, sizeof(req)) < 0) {
+            close(s);
+            vmsg(errmsg);
+            return FULLUPDATE;
+        }
+        close(s);
+    }
+#endif
+
     {
 	/* build tail first. */
 	char tail[STRLEN];