import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public List<MediaItem> search(String username, String text, int width, int height, long publicationDate, int limit, int offset, String type) {
    BasicDBObject q = new BasicDBObject();
    if (!StringUtils.isEmpty(text))
        q.put("description", java.util.regex.Pattern.compile(text, Pattern.CASE_INSENSITIVE));
    if (!StringUtils.isEmpty(type))
        q.put("type", type);
    if (!StringUtils.isEmpty(username)){
        DBObject o = new BasicDBObject();
        o.put("userid",username);
        List<StreamUser> users = userDAO.getStreamUsers(o);
        if(users!=null && users.size()>0){
            q.put("uid",users.get(0).getId());
        }else{
            return new ArrayList<MediaItem>();
        }
    }
    if (width > 0)
        q.put("width", new BasicDBObject("$gt", width));
    if (height > 0)
        q.put("height", new BasicDBObject("$gt", height));
    if (publicationDate > 0)
        q.put("publicationTime", new BasicDBObject("$gt", publicationDate));
    return get(q, offset, limit);
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
private List<MediaItem> get(BasicDBObject object, int offset, int limit) {
    DBCursor cursor = dbCollection.find(object).skip(offset);
    List<String> jsonResults = new ArrayList<String>();
    if (limit > 0) {
        cursor = cursor.limit(limit);
    }
    try {
        while (cursor.hasNext()) {
            DBObject current = cursor.next();
            jsonResults.add(JSON.serialize(current));
        }
    } finally {
        cursor.close();
    }
    List<MediaItem> mediaItems = new ArrayList<MediaItem>(jsonResults.size());
    for (String json : jsonResults) {
        MediaItem item = gson.fromJson(json, MediaItem.class);
        StreamUser user = userDAO.getStreamUser(item.getUserId());
        item.setUser(user);
        mediaItems.add(item);
    }
    return mediaItems;
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUser(String uid) {
	
	People peopleSrv = plusSrv.people();
	try {
		Get getRequest = peopleSrv.get(uid);
		Person person = getRequest.execute();
		
		StreamUser streamUser = new GooglePlusStreamUser(person);
		
		return streamUser;
	} catch (IOException e) {
		logger.error("Exception for user " + uid);
	}
	
	return null;
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public MediaItem getItem(String id){
    String json = mongoHandler.findOne("id", id);
    MediaItem item = gson.fromJson(json, MediaItem.class);
    if(item!=null) {
        StreamUser user = userDAO.getStreamUser(item.getUserId());
        item.setUser(user);
    }
    return item;
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
/**
   * Function that returns the StreamUser associated to the MediaItem
   * @param id the StreamUser id 
   * @return the StreamUser that has the specified id
   * @throws Exception 
   */
  public static StreamUser getStreamUser(String id) throws Exception{
  	
  	StreamUserDAOImpl dao = new StreamUserDAOImpl("160.40.50.242", db, collection);
StreamUser su = dao.getStreamUser(id);

return su;
  }
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public void updateStreamUserPopularity(StreamUser user) {
    String description = user.getDescription();
    if (description != null) {
        UpdateItem changes = new UpdateItem();
        changes.setField("description", description);
        mongoHandler.update("id", user.getId(), changes);
    }
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public void updateStreamUserStatistics(StreamUser user) {
    if (user != null) {
    	DBObject incs = new BasicDBObject();
    	boolean update = false;
    	if(user.getShares()>0) {
    		update = true;
    		incs.put("shares", user.getShares());
    	}
    	if(user.getMentions()>0) {
    		update = true;
    		incs.put("mentions", user.getMentions());
    	}
    	
    	if(user.getItems()>0) {
    		update = true;
    		incs.put("items", user.getItems());
    	}
    	
		BasicDBObject sets = new BasicDBObject();
		if(user.getImageUrl() != null)
			sets.put("imageUrl", user.getImageUrl());
		if(user.getProfileImage() != null)
			sets.put("profileImage", user.getProfileImage());
		if(user.getName() != null) 
			sets.put("name", user.getName());
		if(user.getLastUpdated() != null && user.getLastUpdated() > 0) 
			sets.put("lastUpdated", user.getLastUpdated());
		
    	if(update) {
    		DBObject change = new BasicDBObject("$inc", incs);
    		if(!sets.isEmpty()) {
    			change.put("$set", sets);
    		}
    		
    		mongoHandler.update("id", user.getId(), change);
    	}
    }
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUserByName(String username) {

    Selector query = new Selector();
    query.select("username", StringEscapeUtils.escapeHtml(username));
    query.select("streamId", "Twitter");

    String json = mongoHandler.findOne(query);

    StreamUser user = ItemFactory.createUser(json);
    return user;
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public Map<String, StreamUser> getStreamUsers(List<String> ids) {
    Map<String, StreamUser> users = new HashMap<String, StreamUser>();

    DBObject query = new BasicDBObject("id", new BasicDBObject("$in", ids));;

    List<String> response = mongoHandler.findMany(query, ids.size());
    for (String json : response) {
        StreamUser user = ItemFactory.createUser(json);
        users.put(user.getId(), user);
    }

    return users;
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public static void main(String... args) {

        StreamUserDAO dao = null;
		try {
			dao = new StreamUserDAOImpl("xxx.xxx.xxx.xxx", "Streams", "StreamUsers");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        StreamUser user = dao.getStreamUserByName("SethMacFarlane");
        System.out.println("done");


    }
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public TwitPicMediaItem(TwitPicImage image) throws Exception {
	super(new URL(urlBase + image.id + "." + image.type));
	
	//Id
	this.setId("Twitpic#"+image.id);
	//SocialNetwork Name
	this.setStreamId("Twitpic");
	//Type 
	this.setType("image");
	//Time of publication
	try {
		Date date = formatter.parse(image.timestamp);
		this.setPublicationTime(date.getTime());
	}
	catch(Exception e){
		
	}
	//PageUrl
	this.setPageUrl(pageBase + image.short_id);
	//Thumbnail
	this.setThumbnail(thumbBase + image.id + "." + image.type);
	//Title
	this.setTitle(image.message);
	//Tags
	if(image.tags != null) {
		this.setTags(image.tags.split(","));
	}
	//Popularity
	comments = new Long(image.number_of_comments);
	views = new Long(image.views);
	//Size
	this.setSize(image.width, image.height);
	
	this.setUserId(image.user_id);
	StreamUser user = new TwitPicStreamUser(image.user);
	this.setUser(user);
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public FlickrItem(Photo photo, StreamUser streamUser) {
	this(photo);

	//User that posted the photo
	this.streamUser = streamUser;
	uid = streamUser.getId();

	for(MediaItem mItem : mediaItems) {
		mItem.setUserId(uid);
	}

}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUser(String uid) {
	try {
		long userId = Long.parseLong(uid);
		User user = twitter.showUser(userId);
		
		StreamUser streamUser = new TwitterStreamUser(user);
		return streamUser;
	}
	catch(Exception e) {
		logger.error(e);
		return null;
	}
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public FlickrRetriever(String flickrKey, String flickrSecret, Integer maxResults, Integer maxRequests, long maxRunningTime) {
	
	this.flickrKey = flickrKey;
	this.flickrSecret = flickrSecret;
	this.maxResults = maxResults;
	this.maxRequests = maxRequests;
	this.maxRunningTime = maxRunningTime;
	
	userMap = new HashMap<String, StreamUser>();
	
	Flickr.debugStream = false;
	
	this.flickr = new Flickr(flickrKey, flickrSecret, new REST());
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUser(String uid) {
	try {
		PeopleInterface peopleInterface = flickr.getPeopleInterface();
		User user = peopleInterface.getInfo(uid);
		
		StreamUser streamUser = new FlickrStreamUser(user);
		return streamUser;
	}
	catch(Exception e) {
		return null;
	}
	
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUser(String uid) {
	try {
		long userId = Long.parseLong(uid);
		UserInfo userInfo = instagram.getUserInfo(userId);
		
		StreamUser user = new InstagramStreamUser(userInfo.getData());
		return user;
	}
	catch(Exception e) {
		logger.error("Exception for user " + uid + " => " + e.getMessage());
		return null;
	}
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
@Override
public StreamUser getStreamUser(String uid) {
	try {
		Page page = facebookClient.fetchObject(uid, Page.class);
		StreamUser facebookUser = new FacebookStreamUser(page);
		return facebookUser;
	}
	catch(Exception e) {
		logger.error(e);
		return null;
	}
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
private void importUsersFromFiles() throws Exception {
    int count=0;
    RevealMediaItemDaoImpl mediaDao = new RevealMediaItemDaoImpl("160.40.51.20", "Showcase", "MediaItems");
    StreamUserDAOImpl userDAO = new StreamUserDAOImpl("160.40.51.20", "Showcase", "StreamUsers");
    BufferedReader reader;
    String jsonFilesFolder = "/home/kandreadou/Pictures/snow/";
    JsonParser parser = new JsonParser();
    List<String> jsonFiles = new ArrayList<String>();
    for (int i = 0; i < 42; i++) {
        jsonFiles.add(jsonFilesFolder + "tweets.json." + i);
    }

    for (int i = 30; i < jsonFiles.size(); i++) {

        System.out.println(jsonFiles.get(i));
        reader = new BufferedReader(new InputStreamReader(
                new FileInputStream(jsonFiles.get(i)), "UTF-8"));
        String line = null;
        while ((line = reader.readLine()) != null) {
            count++;
            if(count==930){
                System.out.println(line);
            }
            try {
                JsonObject tweet = parser.parse(line).getAsJsonObject();
                String tweetId = tweet.get("id").getAsString();
                JsonObject user = tweet.get("user").getAsJsonObject();
                if(user==null){
                    System.out.println("USER IS NULL");
                }
                if (user != null) {
                    String userId = user.get("id").getAsString();
                    StreamUser su = new StreamUser(tweetId, StreamUser.Operation.UPDATE);
                    if (!user.get("description").isJsonNull())
                        su.setDescription(user.get("description").getAsString());
                    su.setId(userId);
                    if (!user.get("url").isJsonNull())
                        su.setPageUrl(user.get("url").getAsString());
                    if (!user.get("name").isJsonNull())
                        su.setName(user.get("name").getAsString());
                    if (!user.get("profile_image_url").isJsonNull())
                        su.setImageUrl(user.get("profile_image_url").getAsString());
                    if (user.has("followers_count"))
                        su.setFollowers(user.get("followers_count").getAsLong());
                    if (!user.get("screen_name").isJsonNull()){
                        String screenName = user.get("screen_name").getAsString();
                        su.setUserid(screenName);
                        su.setUrl("https://twitter.com/"+screenName);
                    }

                    MediaItem item = mediaDao.getMediaItem(tweetId);

                    if (item != null) {
                        //System.out.println(item);
                        item.setUserId(userId);
                        //System.out.println(user);
                        //mediaDao.updateMediaItem(item);
                        if (!userDAO.exists(su.getId()))
                            userDAO.insertStreamUser(su);
                    }
                }
            } catch (Exception ex) {
                System.out.println(ex +"line "+line);
            }
        }
        reader.close();
    }
}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public static String getMediaItemsURL(String url) {
		SocialMediaRetriever retriever = null;
		String mediaId = null;
		String source = null;
		Matcher matcher;
		if((matcher = instagramPattern.matcher(url)).matches()) {
			mediaId = matcher.group(1);
			retriever = retrievers.get("instagram");
			source = "instagram";
		}
		else if((matcher = youtubePattern.matcher(url)).matches()) {
			mediaId = matcher.group(1);
			retriever = retrievers.get("youtube");
			source = "youtube";
		}
		else if((matcher = vimeoPattern.matcher(url)).matches()){
			mediaId = matcher.group(1);
			retriever = retrievers.get("vimeo");
			source = "vimeo";
		}
		else if((matcher = twitpicPattern.matcher(url)).matches()) {
			mediaId = matcher.group(1);
			retriever = retrievers.get("twitpic");
			source = "twitpic";
		}
		else if((matcher = dailymotionPattern.matcher(url)).matches()) {
			mediaId = matcher.group(1);
			retriever = retrievers.get("dailymotion");
			source = "dailymotion";
		}
		else if((matcher = facebookPattern.matcher(url)).matches()) {
			mediaId = matcher.group(1);
			retriever = retrievers.get("facebook");
			source = "facebook";
		}
		else {
//			logger.error(url + " matches nothing.");
			return null;
		}

		if(mediaId == null || retriever == null) {
			return null;
		}

		try {
			MediaItem mediaItem = retriever.getMediaItem(mediaId);
			if(mediaItem == null) {
//				logger.info(mediaId + " from " + source + " is null");
				return null;
			}

			mediaItem.setPageUrl(url);

			StreamUser streamUser = mediaItem.getUser();
			String userid = mediaItem.getUserId();

                        /*
			if(streamUser == null || userid == null) {
				streamUser = retriever.getStreamUser(userid);
				if(streamUser == null) {
					throw new Exception("Missing " + mediaItem.getStreamId() + " user: " + userid);
				}
				mediaItem.setUser(streamUser);
				mediaItem.setUserId(streamUser.getId());
			}*/
                        if(mediaItem==null)
                            return null;
                        else
                        {   
                            String res=mediaItem.getUrl();
                            if(res==null) res="empty";
                            return res;
                        }
		}
		catch(Exception e) {
			//logger.error(e);
                    return null;
		}
	}
 

import eu.socialsensor.framework.common.domain.StreamUser; //导入依赖的package包/类
public void execute(Tuple tuple) {
	if(_mediaItemDAO != null) {
		try {
			
		if(++received%1000==0) {
			_logger.info(received + " media items received. " + newMedia + " are new and " + existedMedia + " already exists!");
		}
			
		MediaItem mediaItem = (MediaItem) tuple.getValueByField("MediaItem");
			if(mediaItem == null)
				return;
		
			String mId = mediaItem.getId();
			
			if(_mediaItemDAO.exists(mId)) {
			
				existedMedia++;
				
				UpdateItem update = new UpdateItem();
				update.setField("vIndexed", mediaItem.isVisualIndexed());
				update.setField("status", mediaItem.isVisualIndexed() ? "indexed" : "failed");
			
				Integer width = mediaItem.getWidth();
				Integer height = mediaItem.getHeight();
				if(width != null && height != null && width != -1 && height != -1) {
					update.setField("height", height);
					update.setField("width", width);
				}
			
				List<Concept> concepts = mediaItem.getConcepts();
				if(concepts != null) {
					update.setField("concepts", concepts);
				}
				
				String clusterId = mediaItem.getClusterId();
				if(clusterId != null) {
					update.setField("clusterId", clusterId);
				}
				
				_mediaItemDAO.updateMediaItem(mediaItem.getId(), update);
			}
			else {
				newMedia++;
				_mediaItemDAO.addMediaItem(mediaItem);
				
				StreamUser user = mediaItem.getUser();
				
				if(user != null && _streamUsersDAO != null) {
					user.setLastUpdated(System.currentTimeMillis());
					
					if(!_streamUsersDAO.exists(user.getId())) {
						_streamUsersDAO.insertStreamUser(user);
					}
				}
			}
			
		}
		catch(Exception e) {
			_logger.error(e);
		}
	}
}