# -*- coding: utf-8 -*-
-from xml.dom import minidom
-from datetime import datetime, timedelta
+from datetime import datetime
import time
import re
import os
import gc
from django.utils.translation import ugettext as _
-from django.template.defaultfilters import slugify
-from forum.models.utils import dbsafe_encode
-from orm import orm
from django.utils.encoding import force_unicode
from xml.sax import make_parser
from xml.sax.handler import ContentHandler
+def create_orm():
+ from django.conf import settings
+ from south.orm import FakeORM
+
+ get_migration_number_re = re.compile(r'^((\d+)_.*)\.py$')
+
+ migrations_folder = os.path.join(settings.SITE_SRC_ROOT, 'forum/migrations')
+
+ highest_number = 0
+ highest_file = None
+
+ for f in os.listdir(migrations_folder):
+ if os.path.isfile(os.path.join(migrations_folder, f)):
+ m = get_migration_number_re.match(f)
+
+ if m:
+ found = int(m.group(2))
+
+ if found > highest_number:
+ highest_number = found
+ highest_file = m.group(1)
+
+ mod = __import__('forum.migrations.%s' % highest_file, globals(), locals(), ['forum.migrations'])
+ return FakeORM(getattr(mod, 'Migration'), "forum")
+
+orm = create_orm()
+
class SXTableHandler(ContentHandler):
def __init__(self, fname, callback):
self.in_row = False
class IdMapper(dict):
+
+ def __init__(self):
+ self.default = 1
+
def __getitem__(self, key):
key = int(key)
- return super(IdMapper, self).get(key, 1)
+ return super(IdMapper, self).get(key, self.default)
def __setitem__(self, key, value):
super(IdMapper, self).__setitem__(int(key), int(value))
openidre = re.compile('^https?\:\/\/')
def userimport(path, options):
-#users = readTable(dump, "Users")
usernames = []
openids = set()
uidmapper = IdMapper()
- #merged_users = []
+ authenticated_user = options.get('authenticated_user', None)
owneruid = options.get('owneruid', None)
#check for empty values
if not owneruid:
owneruid = None
+ else:
+ owneruid = int(owneruid)
def callback(sxu):
create = True
+ set_mapper_defaults = False
if sxu.get('id') == '-1':
return
#print "\n".join(["%s : %s" % i for i in sxu.items()])
- if int(sxu.get('id')) == int(owneruid):
- osqau = orm.User.objects.get(id=1)
- for assoc in orm.AuthKeyUserAssociation.objects.filter(user=osqau):
- openids.add(assoc.key)
- uidmapper[owneruid] = 1
- uidmapper[-1] = 1
- create = False
- else:
+
+ if (owneruid and (int(sxu.get('id')) == owneruid)) or (
+ (not owneruid) and len(uidmapper)):
+
+ set_mapper_defaults = True
+
+ if authenticated_user:
+ osqau = orm.User.objects.get(id=authenticated_user.id)
+
+ for assoc in orm.AuthKeyUserAssociation.objects.filter(user=osqau):
+ openids.add(assoc.key)
+
+ uidmapper[owneruid] = osqau.id
+ create = False
+
+ sxbadges = sxu.get('badgesummary', None)
+ badges = {'1':'0', '2':'0', '3':'0'}
+
+ if sxbadges:
+ badges.update(dict([b.split('=') for b in sxbadges.split()]))
+
+ if create:
username = unicode(sxu.get('displayname',
sxu.get('displaynamecleaned', sxu.get('realname', final_username_attempt(sxu)))))[:30]
if not totest in usernames:
username = totest
- break
+ break
- sxbadges = sxu.get('badgesummary', None)
- badges = {'1':'0', '2':'0', '3':'0'}
-
- if sxbadges:
- badges.update(dict([b.split('=') for b in sxbadges.split()]))
-
- if create:
osqau = orm.User(
id = sxu.get('id'),
username = username,
#merged_users.append(osqau.id)
osqau.save()
+ if set_mapper_defaults:
+ uidmapper[-1] = osqau.id
+ uidmapper.default = osqau.id
+
usernames.append(osqau.username)
openid = sxu.get('openid', None)
readTable(path, "Users", callback)
- if uidmapper[-1] == -1:
- uidmapper[-1] = 1
+ #if uidmapper[-1] == -1:
+ # uidmapper[-1] = 1
return uidmapper
def tagsimport(dump, uidmap):
-#tags = readTable(dump, "Tags")
tagmap = {}
post.state_string = "".join("(%s)" % s for s in re.findall('\w+', post.state_string) if s != name)
def postimport(dump, uidmap, tagmap):
-#history = {}
-#accepted = {}
- all = []
-
- #for h in readTable(dump, "PostHistory"):
- # if not history.get(h.get('postid'), None):
- # history[h.get('postid')] = []
- #
- # history[h.get('postid')].append(h)
-
- #posts = readTable(dump, "Posts")
+ all = {}
def callback(sxpost):
nodetype = (sxpost.get('posttypeid') == '1') and "nodetype" or "answer"
post.extra_count = sxpost.get('viewcount', 0)
add_tags_to_post(post, tagmap)
+ all[int(post.id)] = int(post.id)
else:
post.parent_id = sxpost['parentid']
+ post.abs_parent_id = sxpost['parentid']
+ all[int(post.id)] = int(sxpost['parentid'])
post.save()
- all.append(int(post.id))
+ create_and_activate_revision(post)
del post
return all
-def comment_import(dump, uidmap, posts):
-#comments = readTable(dump, "PostComments")
+def comment_import(dump, uidmap, absparent_map):
+ posts = absparent_map.keys()
+
currid = IdIncrementer(max(posts))
mapping = {}
author_id = uidmap[sxc.get('userid', 1)],
body = sxc['text'],
parent_id = sxc.get('postid'),
+ abs_parent_id = absparent_map.get(int(sxc.get('postid')), sxc.get('postid'))
)
if sxc.get('deletiondate', None):
action_date = oc.added_at
)
+ create_and_activate_revision(oc)
+
create_action.save()
oc.save()
tags = [tag for tag in [tagmap.get(name.strip()) for name in post.tagnames.split(u' ') if name] if tag]
post.tagnames = " ".join([t.name for t in tags]).strip()
post.tags = tags
- create_and_activate_revision(post)
def create_and_activate_revision(post):
post.save()
def post_vote_import(dump, uidmap, posts):
-#votes = readTable(dump, "Posts2Votes")
close_reasons = {}
def close_callback(r):
def comment_vote_import(dump, uidmap, comments):
-#votes = readTable(dump, "Comments2Votes")
user2vote = []
comments2score = {}
def badges_import(dump, uidmap, post_list):
-#node_ctype = orm['contenttypes.contenttype'].objects.get(name='node')
sxbadges = {}
osqaa.save()
badge.awarded_count += 1
+
user_badge_count[user_id] += 1
readTable(dump, "Users2Badges", callback)
for badge in obadges.values():
badge.save()
-def pages_import(dump, currid):
+def save_setting(k, v):
+ try:
+ kv = orm.KeyValue.objects.get(key=k)
+ kv.value = v
+ except:
+ kv = orm.KeyValue(key = k, value = v)
+
+ kv.save()
+
+
+def pages_import(dump, currid, owner):
currid = IdIncrementer(currid)
registry = {}
- #sx_pages = readTable(dump, "FlatPages")
def callback(sxp):
currid.inc()
'sidebar_render': "html",
'comments': False
}),
- author_id = 1
+ author_id = owner
)
+ create_and_activate_revision(page)
+
page.save()
registry[sxp['url'][1:]] = page.id
readTable(dump, "FlatPages", callback)
- kv = orm.KeyValue(key='STATIC_PAGE_REGISTRY', value=dbsafe_encode(registry))
- kv.save()
+ save_setting('STATIC_PAGE_REGISTRY', dbsafe_encode(registry))
sx2osqa_set_map = {
u'theme.html.name': 'APP_TITLE',
def static_import(dump):
-#sx_sets = readTable(dump, "ThemeTextResources")
sx_unknown = {}
def callback(set):
if unicode(set['name']) in sx2osqa_set_map:
- try:
- kv = orm.KeyValue.objects.get(key=sx2osqa_set_map[set['name']])
- kv.value = dbsafe_encode(html_decode(set['value']))
- except:
- kv = orm.KeyValue(
- key = sx2osqa_set_map[set['name']],
- value = dbsafe_encode(html_decode(set['value']))
- )
-
- kv.save()
+ save_setting(sx2osqa_set_map[set['name']], dbsafe_encode(html_decode(set['value'])))
else:
sx_unknown[set['name']] = html_decode(set['value'])
readTable(dump, "ThemeTextResources", callback)
- unknown = orm.KeyValue(key='SXIMPORT_UNKNOWN_SETS', value=dbsafe_encode(sx_unknown))
- unknown.save()
+ save_setting('SXIMPORT_UNKNOWN_SETS', dbsafe_encode(sx_unknown))
def disable_triggers():
from south.db import db
badges_import(dump, uidmap, posts)
- pages_import(dump, max(posts))
+ pages_import(dump, max(posts), uidmap.default)
static_import(dump)
gc.collect()
projects / osqa.git / blobdiff